add tenant cleanup methods for in-memory structs on tenant deletion (#1591)

1. deregister the schema for a tenant on deletion
2. remove in-memory dashboards for a tenant
3. remove in-memory filters for a tenant
4. remove in-memory alerts for a tenant

Author: nikhilsinhaparseable

src/alerts/alert_traits.rs

@@ -119,6 +119,7 @@ pub trait AlertManagerTrait: Send + Sync {
         &self,
         tenant_id: &Option<String>,
     ) -> HashMap<Ulid, Box<dyn AlertTrait>>;
+    async fn delete_all_for_tenant(&self, tenant_id: &str);
 }
 
 #[async_trait]

src/alerts/mod.rs

@@ -1484,6 +1484,31 @@ impl AlertManagerTrait for Alerts {
         // let alerts = self.alerts.read().await;
         // alerts.iter().map(|(k, v)| (*k, v.clone_box())).collect()
     }
+
+    async fn delete_all_for_tenant(&self, tenant_id: &str) {
+        let tenant = if tenant_id.is_empty() {
+            DEFAULT_TENANT
+        } else {
+            tenant_id
+        };
+
+        let alert_ids: Vec<Ulid> = {
+            let read_access = self.alerts.read().await;
+            if let Some(alerts) = read_access.get(tenant) {
+                alerts.keys().copied().collect()
+            } else {
+                return;
+            }
+        };
+
+        for alert_id in &alert_ids {
+            if let Err(e) = self.sender.send(AlertTask::Delete(*alert_id)).await {
+                warn!("Failed to cancel alert task {alert_id} for tenant {tenant}: {e}");
+            }
+        }
+
+        self.alerts.write().await.remove(tenant);
+    }
 }
 
 // TODO: add RBAC

src/query/mod.rs

@@ -118,6 +118,17 @@ impl InMemorySessionContext {
             )
             .expect("Should be able to register new schema");
     }
+
+    pub fn remove_schema(&self, tenant_id: &str) {
+        if let Some(catalog) = self
+            .session_context
+            .write()
+            .expect("SessionContext should be writeable")
+            .catalog("datafusion")
+        {
+            let _ = catalog.deregister_schema(tenant_id, true);
+        }
+    }
 }
 
 /// This function executes a query on the dedicated runtime, ensuring that the query is not isolated to a single thread/CPU

src/users/dashboards.rs

@@ -505,6 +505,10 @@ impl Dashboards {
         }
     }
 
+    pub async fn delete_tenant(&self, tenant_id: &str) {
+        self.0.write().await.remove(tenant_id);
+    }
+
     /// Ensure the user is the owner of the dashboard
     /// This function is called when updating or deleting a dashboard
     /// check if the user is the owner of the dashboard

src/users/filters.rs

@@ -164,6 +164,10 @@ impl Filters {
         }
     }
 
+    pub async fn delete_tenant(&self, tenant_id: &str) {
+        self.0.write().await.remove(tenant_id);
+    }
+
     pub async fn list_filters(&self, key: &SessionKey) -> Vec<Filter> {
         let read = self.0.read().await;
         let tenant_id = get_tenant_id_from_key(key);