fix: remove associated filters during stream deletion

src/handlers/http/logstream.rs

@@ -29,6 +29,7 @@ use crate::rbac::role::Action;
 use crate::stats::{Stats, event_labels_date, storage_size_labels_date};
 use crate::storage::retention::Retention;
 use crate::storage::{ObjectStoreFormat, StreamInfo, StreamType};
+use crate::users::filters::{FILTERS, Filter};
 use crate::utils::actix::extract_session_key_from_req;
 use crate::utils::json::flatten::{
     self, convert_to_array, generic_flattening, has_more_than_max_allowed_levels,
@@ -56,6 +57,21 @@ pub async fn delete(stream_name: Path<String>) -> Result<impl Responder, StreamE
 
     let objectstore = PARSEABLE.storage.get_object_store();
 
+    let all_filters = PARSEABLE.metastore.get_filters().await?;
+    // collect filters associated with the logstream being deleted
+    let filters_for_stream: Vec<Filter> = all_filters
+        .into_iter()
+        .filter(|filter| filter.stream_name == stream_name)
+        .collect();
+
+    for filter in filters_for_stream.iter() {
+        PARSEABLE.metastore.delete_filter(filter).await?;
+
+        if let Some(filter_id) = filter.filter_id.as_ref() {
+            FILTERS.delete_filter(filter_id).await;
+        }
+    }
+
     // Delete from storage
     objectstore.delete_stream(&stream_name).await?;
     // Delete from staging