diff --git a/src/infra/src/pipeline/mysql.rs b/src/infra/src/pipeline/mysql.rs index 48f5728128..2517fa644a 100644 --- a/src/infra/src/pipeline/mysql.rs +++ b/src/infra/src/pipeline/mysql.rs @@ -271,18 +271,11 @@ SELECT * FROM pipeline WHERE org = ? AND source_type = ? AND stream_org = ? AND async fn get_by_id(&self, pipeline_id: &str) -> Result { let pool = CLIENT_RO.clone(); let query = r#"SELECT * FROM pipeline WHERE id = ?;"#; - let pipeline = match sqlx::query_as::<_, Pipeline>(query) + sqlx::query_as::<_, Pipeline>(query) .bind(pipeline_id) .fetch_one(&pool) .await - { - Ok(pipeline) => pipeline, - Err(e) => { - log::debug!("[MYSQL] get pipeline by id error: {}", e); - return Err(Error::from(DbError::KeyNotExists(pipeline_id.to_string()))); - } - }; - Ok(pipeline) + .map_err(|_| Error::from(DbError::KeyNotExists(pipeline_id.to_string()))) } async fn get_with_same_source_stream(&self, pipeline: &Pipeline) -> Result { diff --git a/src/infra/src/pipeline/postgres.rs b/src/infra/src/pipeline/postgres.rs index 1e2c194b2a..233b4882a2 100644 --- a/src/infra/src/pipeline/postgres.rs +++ b/src/infra/src/pipeline/postgres.rs @@ -268,18 +268,11 @@ SELECT * FROM pipeline WHERE org = $1 AND source_type = $2 AND stream_org = $3 A async fn get_by_id(&self, pipeline_id: &str) -> Result { let pool = CLIENT_RO.clone(); let query = "SELECT * FROM pipeline WHERE id = $1;"; - let pipeline = match sqlx::query_as::<_, Pipeline>(query) + sqlx::query_as::<_, Pipeline>(query) .bind(pipeline_id) .fetch_one(&pool) .await - { - Ok(pipeline) => pipeline, - Err(e) => { - log::error!("[POSTGRES] get pipeline by id error: {}", e); - return Err(Error::from(DbError::KeyNotExists(pipeline_id.to_string()))); - } - }; - Ok(pipeline) + .map_err(|_| Error::from(DbError::KeyNotExists(pipeline_id.to_string()))) } async fn get_with_same_source_stream(&self, pipeline: &Pipeline) -> Result { diff --git a/src/infra/src/pipeline/sqlite.rs b/src/infra/src/pipeline/sqlite.rs index bf12b05cf8..340f0f110c 100644 --- a/src/infra/src/pipeline/sqlite.rs +++ b/src/infra/src/pipeline/sqlite.rs @@ -270,18 +270,11 @@ SELECT * FROM pipeline WHERE org = $1 AND source_type = $2 AND stream_org = $3 A async fn get_by_id(&self, pipeline_id: &str) -> Result { let pool = CLIENT_RO.clone(); let query = "SELECT * FROM pipeline WHERE id = $1;"; - let pipeline = match sqlx::query_as::<_, Pipeline>(query) + sqlx::query_as::<_, Pipeline>(query) .bind(pipeline_id) .fetch_one(&pool) .await - { - Ok(pipeline) => pipeline, - Err(e) => { - log::error!("[SQLITE] get pipeline by id error: {}", e); - return Err(Error::from(DbError::KeyNotExists(pipeline_id.to_string()))); - } - }; - Ok(pipeline) + .map_err(|_| Error::from(DbError::KeyNotExists(pipeline_id.to_string()))) } async fn get_with_same_source_stream(&self, pipeline: &Pipeline) -> Result { diff --git a/src/service/alerts/destinations.rs b/src/service/alerts/destinations.rs index fdef23f695..8885bbc641 100644 --- a/src/service/alerts/destinations.rs +++ b/src/service/alerts/destinations.rs @@ -146,7 +146,9 @@ pub async fn list( pub async fn delete(org_id: &str, name: &str) -> Result<(), DestinationError> { let cacher = ALERTS.read().await; for (stream_key, (_, alert)) in cacher.iter() { - if stream_key.starts_with(&format!("{}/", org_id)) && alert.destinations.contains(&name.to_string()) { + if stream_key.starts_with(&format!("{}/", org_id)) + && alert.destinations.contains(&name.to_string()) + { return Err(DestinationError::UsedByAlert(alert.name.to_string())); } }