Cleanup appservice events after removing the appservice

src/database/admin.rs

@@ -112,7 +112,18 @@ impl Admin {
                                 guard.appservice.register_appservice(yaml).unwrap(); // TODO handle error
                             }
                             AdminCommand::UnregisterAppservice(service_name) => {
-                                guard.appservice.unregister_appservice(&service_name).unwrap(); // TODO: see above
+                                if let Ok(_) = guard.appservice.unregister_appservice(&service_name) {
+                                    if let Ok(_) = guard.sending.cleanup_events(&service_name) {
+                                        let msg: String = format!("OK. Appservice {} removed", service_name);
+                                        send_message(RoomMessageEventContent::text_plain(msg), guard, &state_lock);
+                                    } else {
+                                        let msg: String = format!("WARN: Appservice {} removed, but failed to cleanup events", service_name);
+                                        send_message(RoomMessageEventContent::text_plain(msg), guard, &state_lock);
+                                    }
+                                } else {
+                                    let msg: String = format!("ERR. Appservice {} not removed", service_name);
+                                    send_message(RoomMessageEventContent::text_plain(msg), guard, &state_lock);
+                                }
                             }
                             AdminCommand::ListAppservices => {
                                 if let Ok(appservices) = guard.appservice.iter_ids().map(|ids| ids.collect::<Vec<_>>()) {

src/database/sending.rs

@@ -480,6 +480,26 @@ impl Sending {
         hash.as_ref().to_owned()
     }
 
+    /// Cleanup event data
+    /// Used for instance after we remove an appservice registration
+    ///
+    #[tracing::instrument(skip(self))]
+    pub fn cleanup_events(&self, key_id: &str) -> Result<()> {
+        let mut prefix = b"+".to_vec();
+        prefix.extend_from_slice(key_id.as_bytes());
+        prefix.push(0xff);
+
+        for (key, _) in self.servercurrentevent_data.scan_prefix(prefix.clone()) {
+            self.servercurrentevent_data.remove(&key).unwrap();
+        }
+
+        for (key, _) in self.servernameevent_data.scan_prefix(prefix.clone()) {
+            self.servernameevent_data.remove(&key).unwrap();
+        }
+
+        Ok(())
+    }
+
     #[tracing::instrument(skip(db, events, kind))]
     async fn handle_events(
         kind: OutgoingKind,
@@ -520,8 +540,15 @@ impl Sending {
                     &db.globals,
                     db.appservice
                         .get_registration(server.as_str())
-                        .unwrap()
-                        .unwrap(), // TODO: handle error
+                        .map_err(|e| (kind.clone(), e))?
+                        .ok_or_else(|| {
+                            (
+                                kind.clone(),
+                                Error::bad_database(
+                                    "[Appservice] Could not load registration from db.",
+                                ),
+                            )
+                        })?,
                     appservice::event::push_events::v1::Request {
                         events: &pdu_jsons,
                         txn_id: (&*base64::encode_config(