fix: not overwrite pipeline_id when imported (#7179)

src/handler/http/request/alerts/deprecated.rs

@@ -63,8 +63,14 @@ pub async fn save_alert(
     path: web::Path<(String, String)>,
     alert: web::Json<Alert>,
     user_email: UserEmail,
+    req: HttpRequest,
 ) -> HttpResponse {
     let (org_id, stream_name) = path.into_inner();
+    let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
+    let overwrite = match query.get("overwrite") {
+        Some(v) => v.parse::<bool>().unwrap_or_default(),
+        None => false,
+    };
 
     // Hack for frequency: convert minutes to seconds
     let mut alert = alert.into_inner();
@@ -77,7 +83,7 @@ pub async fn save_alert(
     alert.set_last_satisfied_at(None);
     alert.set_last_triggered_at(None);
 
-    match alert::save(&org_id, &stream_name, "", alert, true).await {
+    match alert::save(&org_id, &stream_name, "", alert, true, overwrite).await {
         Ok(_) => MetaHttpResponse::ok("Alert saved"),
         Err(e) => e.into(),
     }
@@ -118,7 +124,7 @@ pub async fn update_alert(
     alert.trigger_condition.frequency *= 60;
     alert.last_edited_by = Some(user_email.user_id);
     alert.updated_at = Some(datetime_now());
-    match alert::save(&org_id, &stream_name, &name, alert, false).await {
+    match alert::save(&org_id, &stream_name, &name, alert, false, false).await {
         Ok(_) => MetaHttpResponse::ok("Alert Updated"),
         Err(e) => e.into(),
     }

src/handler/http/request/alerts/mod.rs

@@ -32,7 +32,7 @@ use crate::{
             },
             responses::{EnableAlertResponseBody, GetAlertResponseBody, ListAlertsResponseBody},
         },
-        request::dashboards::get_folder,
+        request::dashboards::{get_folder, is_overwrite},
     },
     service::{
         alerts::alert::{self, AlertError},
@@ -111,7 +111,9 @@ pub async fn create_alert(
     let org_id = path.into_inner();
     let req_body = req_body.into_inner();
 
-    let folder_id = get_folder(req);
+    let query_str = req.query_string();
+    let folder_id = get_folder(query_str);
+    let overwrite = is_overwrite(query_str);
     let mut alert: MetaAlert = req_body.into();
     if alert.owner.clone().filter(|o| !o.is_empty()).is_none() {
         alert.owner = Some(user_email.user_id.clone());
@@ -119,7 +121,7 @@ pub async fn create_alert(
     alert.last_edited_by = Some(user_email.user_id);
 
     let client = ORM_CLIENT.get_or_init(connect_to_orm).await;
-    match alert::create(client, &org_id, &folder_id, alert).await {
+    match alert::create(client, &org_id, &folder_id, alert, overwrite).await {
         Ok(v) => MetaHttpResponse::json(
             MetaHttpResponse::message(StatusCode::OK, "Alert saved")
                 .with_id(v.id.map(|id| id.to_string()).unwrap_or_default())

src/handler/http/request/dashboards/mod.rs

@@ -102,7 +102,7 @@ pub async fn create_dashboard(
     user_email: UserEmail,
 ) -> impl Responder {
     let org_id = path.into_inner();
-    let folder = get_folder(req);
+    let folder = get_folder(req.query_string());
     let mut dashboard: config::meta::dashboards::Dashboard = match req_body.into_inner().try_into()
     {
         Ok(dashboard) => dashboard,
@@ -372,11 +372,19 @@ async fn move_dashboards(
     }
 }
 
-pub fn get_folder(req: HttpRequest) -> String {
-    let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
+pub fn get_folder(query_str: &str) -> String {
+    let query = web::Query::<HashMap<String, String>>::from_query(query_str).unwrap();
     crate::common::utils::http::get_folder(&query)
 }
 
+pub fn is_overwrite(query_str: &str) -> bool {
+    let query = web::Query::<HashMap<String, String>>::from_query(query_str).unwrap();
+    match query.get("overwrite") {
+        Some(v) => v.parse::<bool>().unwrap_or_default(),
+        None => false,
+    }
+}
+
 /// Tries to get the user ID from the request headers.
 fn get_user_id(req: HttpRequest) -> Option<String> {
     req.headers()

src/handler/http/request/pipeline.rs

@@ -59,13 +59,20 @@ impl From<PipelineError> for HttpResponse {
 pub async fn save_pipeline(
     path: web::Path<String>,
     pipeline: web::Json<Pipeline>,
-    _req: HttpRequest,
+    req: HttpRequest,
 ) -> Result<HttpResponse, Error> {
     let org_id = path.into_inner();
     let mut pipeline = pipeline.into_inner();
     pipeline.name = pipeline.name.trim().to_lowercase();
     pipeline.org = org_id;
-    pipeline.id = ider::generate();
+    let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
+    let overwrite = match query.get("overwrite") {
+        Some(v) => v.parse::<bool>().unwrap_or_default(),
+        None => false,
+    };
+    if !overwrite {
+        pipeline.id = ider::generate();
+    }
     match pipeline::save_pipeline(pipeline).await {
         Ok(()) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message(
             http::StatusCode::OK,
@@ -158,10 +165,7 @@ async fn list_pipelines(
     )
 )]
 #[get("/{org_id}/pipelines/streams")]
-async fn list_streams_with_pipeline(
-    path: web::Path<String>,
-    _req: HttpRequest,
-) -> Result<HttpResponse, Error> {
+async fn list_streams_with_pipeline(path: web::Path<String>) -> Result<HttpResponse, Error> {
     let org_id = path.into_inner();
     match pipeline::list_streams_with_pipeline(&org_id).await {
         Ok(stream_params) => Ok(HttpResponse::Ok().json(stream_params)),
@@ -189,10 +193,7 @@ async fn list_streams_with_pipeline(
     )
 )]
 #[delete("/{org_id}/pipelines/{pipeline_id}")]
-async fn delete_pipeline(
-    path: web::Path<(String, String)>,
-    _req: HttpRequest,
-) -> Result<HttpResponse, Error> {
+async fn delete_pipeline(path: web::Path<(String, String)>) -> Result<HttpResponse, Error> {
     let (_org_id, pipeline_id) = path.into_inner();
     match pipeline::delete_pipeline(&pipeline_id).await {
         Ok(()) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message(
@@ -223,10 +224,7 @@ async fn delete_pipeline(
     )
 )]
 #[put("/{org_id}/pipelines")]
-pub async fn update_pipeline(
-    pipeline: web::Json<Pipeline>,
-    _req: HttpRequest,
-) -> Result<HttpResponse, Error> {
+pub async fn update_pipeline(pipeline: web::Json<Pipeline>) -> Result<HttpResponse, Error> {
     let pipeline = pipeline.into_inner();
     match pipeline::update_pipeline(pipeline).await {
         Ok(()) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message(

src/service/alerts/alert.rs

@@ -190,6 +190,7 @@ pub async fn save(
     name: &str,
     mut alert: Alert,
     create: bool,
+    overwrite: bool,
 ) -> Result<(), AlertError> {
     // Currently all alerts are stored in the default folder so create the
     // default folder for the org if it doesn't exist yet.
@@ -197,7 +198,7 @@ pub async fn save(
         create_default_alerts_folder(org_id).await?;
     };
 
-    prepare_alert(org_id, stream_name, name, &mut alert, create).await?;
+    prepare_alert(org_id, stream_name, name, &mut alert, create, overwrite).await?;
 
     // save the alert
     // TODO: Get the folder id
@@ -239,6 +240,7 @@ async fn prepare_alert(
     name: &str,
     alert: &mut Alert,
     create: bool,
+    overwrite: bool,
 ) -> Result<(), AlertError> {
     if !name.is_empty() {
         alert.name = name.to_string();
@@ -261,7 +263,7 @@ async fn prepare_alert(
     if let Some(alert_id) = alert.id {
         match get_by_id_db(org_id, alert_id).await {
             Ok(old_alert) => {
-                if create {
+                if create && !overwrite {
                     return Err(AlertError::CreateAlreadyExists);
                 }
                 alert.owner = old_alert.owner;
@@ -459,6 +461,7 @@ pub async fn create<C: TransactionTrait>(
     org_id: &str,
     folder_id: &str,
     mut alert: Alert,
+    overwrite: bool,
 ) -> Result<Alert, AlertError> {
     if !table::folders::exists(org_id, folder_id, FolderType::Alerts).await? {
         if folder_id == DEFAULT_FOLDER {
@@ -470,9 +473,17 @@ pub async fn create<C: TransactionTrait>(
 
     let alert_name = alert.name.clone();
     let stream_name = alert.stream_name.clone();
-    prepare_alert(org_id, &stream_name, &alert_name, &mut alert, true).await?;
+    prepare_alert(
+        org_id,
+        &stream_name,
+        &alert_name,
+        &mut alert,
+        true,
+        overwrite,
+    )
+    .await?;
 
-    let alert = db::alerts::alert::create(conn, org_id, folder_id, alert).await?;
+    let alert = db::alerts::alert::create(conn, org_id, folder_id, alert, overwrite).await?;
 
     set_ownership(
         org_id,
@@ -578,7 +589,7 @@ pub async fn update<C: ConnectionTrait + TransactionTrait>(
     let alert_name = alert.name.clone();
     let stream_name = alert.stream_name.clone();
 
-    prepare_alert(org_id, &stream_name, &alert_name, &mut alert, false).await?;
+    prepare_alert(org_id, &stream_name, &alert_name, &mut alert, false, false).await?;
 
     let alert = db::alerts::alert::update(conn, org_id, dst_folder_id_info, alert).await?;
     #[cfg(feature = "enterprise")]
@@ -1840,7 +1851,7 @@ mod tests {
         let alert_name = "abc/alert";
         let mut alert: Alert = Default::default();
         alert.name = alert_name.to_string();
-        let ret = save(org_id, stream_name, alert_name, alert, true).await;
+        let ret = save(org_id, stream_name, alert_name, alert, true, false).await;
         // alert name should not contain /
         assert!(ret.is_err());
     }

src/service/db/alerts/alert.rs

@@ -150,8 +150,9 @@ pub async fn create<C: TransactionTrait>(
     org_id: &str,
     folder_id: &str,
     alert: Alert,
+    overwrite: bool,
 ) -> Result<Alert, infra::errors::Error> {
-    let alert = table::create(conn, org_id, folder_id, alert, false).await?;
+    let alert = table::create(conn, org_id, folder_id, alert, overwrite).await?;
     let schedule_key = scheduler_key(alert.id);
 
     cluster::emit_put_event(org_id, &alert, Some(folder_id.to_string())).await?;

src/service/pipeline/mod.rs

@@ -33,6 +33,12 @@ pub mod batch_execution;
 
 #[tracing::instrument(skip(pipeline))]
 pub async fn save_pipeline(mut pipeline: Pipeline) -> Result<(), PipelineError> {
+    // check if id is missing
+    if pipeline.id.is_empty() {
+        return Err(PipelineError::InvalidPipeline(
+            "Missing pipeline ID".to_string(),
+        ));
+    }
     // check if another realtime pipeline with the same source stream already exists
     if let PipelineSource::Realtime(stream) = &pipeline.source
         && pipeline::list_streams_with_pipeline(&pipeline.org)