Work on event_handler, lazy_loading, metadata, outlier, pdu_metadata

src/database/key_value.rs

@@ -509,3 +509,142 @@ fn parse_presence_event(bytes: &[u8]) -> Result<PresenceEvent> {
             .map(|timestamp| current_timestamp - timestamp);
     }
 }
+
+impl service::room::lazy_load::Data for KeyValueDatabase {
+    fn lazy_load_was_sent_before(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+        ll_user: &UserId,
+    ) -> Result<bool> {
+        let mut key = user_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(device_id.as_bytes());
+        key.push(0xff);
+        key.extend_from_slice(room_id.as_bytes());
+        key.push(0xff);
+        key.extend_from_slice(ll_user.as_bytes());
+        Ok(self.lazyloadedids.get(&key)?.is_some())
+    }
+
+    fn lazy_load_confirm_delivery(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+        since: u64,
+    ) -> Result<()> {
+        if let Some(user_ids) = self.lazy_load_waiting.lock().unwrap().remove(&(
+            user_id.to_owned(),
+            device_id.to_owned(),
+            room_id.to_owned(),
+            since,
+        )) {
+            let mut prefix = user_id.as_bytes().to_vec();
+            prefix.push(0xff);
+            prefix.extend_from_slice(device_id.as_bytes());
+            prefix.push(0xff);
+            prefix.extend_from_slice(room_id.as_bytes());
+            prefix.push(0xff);
+
+            for ll_id in user_ids {
+                let mut key = prefix.clone();
+                key.extend_from_slice(ll_id.as_bytes());
+                self.lazyloadedids.insert(&key, &[])?;
+            }
+        }
+
+        Ok(())
+    }
+
+    fn lazy_load_reset(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+    ) -> Result<()> {
+        let mut prefix = user_id.as_bytes().to_vec();
+        prefix.push(0xff);
+        prefix.extend_from_slice(device_id.as_bytes());
+        prefix.push(0xff);
+        prefix.extend_from_slice(room_id.as_bytes());
+        prefix.push(0xff);
+
+        for (key, _) in self.lazyloadedids.scan_prefix(prefix) {
+            self.lazyloadedids.remove(&key)?;
+        }
+
+        Ok(())
+    }
+}
+
+impl service::room::metadata::Data for KeyValueDatabase {
+    fn exists(&self, room_id: &RoomId) -> Result<bool> {
+        let prefix = match self.get_shortroomid(room_id)? {
+            Some(b) => b.to_be_bytes().to_vec(),
+            None => return Ok(false),
+        };
+
+        // Look for PDUs in that room.
+        Ok(self
+            .pduid_pdu
+            .iter_from(&prefix, false)
+            .next()
+            .filter(|(k, _)| k.starts_with(&prefix))
+            .is_some())
+    }
+}
+
+impl service::room::outlier::Data for KeyValueDatabase {
+    fn get_outlier_pdu_json(&self, event_id: &EventId) -> Result<Option<CanonicalJsonObject>> {
+        self.eventid_outlierpdu
+            .get(event_id.as_bytes())?
+            .map_or(Ok(None), |pdu| {
+                serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
+            })
+    }
+
+    fn get_outlier_pdu(&self, event_id: &EventId) -> Result<Option<PduEvent>> {
+        self.eventid_outlierpdu
+            .get(event_id.as_bytes())?
+            .map_or(Ok(None), |pdu| {
+                serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
+            })
+    }
+
+    fn add_pdu_outlier(&self, event_id: &EventId, pdu: &CanonicalJsonObject) -> Result<()> {
+        self.eventid_outlierpdu.insert(
+            event_id.as_bytes(),
+            &serde_json::to_vec(&pdu).expect("CanonicalJsonObject is valid"),
+        )
+    }
+}
+
+impl service::room::pdu_metadata::Data for KeyValueDatabase {
+    fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()> {
+        for prev in event_ids {
+            let mut key = room_id.as_bytes().to_vec();
+            key.extend_from_slice(prev.as_bytes());
+            self.referencedevents.insert(&key, &[])?;
+        }
+
+        Ok(())
+    }
+
+    fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> Result<bool> {
+        let mut key = room_id.as_bytes().to_vec();
+        key.extend_from_slice(event_id.as_bytes());
+        Ok(self.referencedevents.get(&key)?.is_some())
+    }
+
+    fn mark_event_soft_failed(&self, event_id: &EventId) -> Result<()> {
+        self.softfailedeventids.insert(event_id.as_bytes(), &[])
+    }
+
+    fn is_event_soft_failed(&self, event_id: &EventId) -> Result<bool> {
+        self.softfailedeventids
+            .get(event_id.as_bytes())
+            .map(|o| o.is_some())
+    }
+}

src/service/rooms/lazy_loading/data.rs

@@ -0,0 +1,24 @@
+pub trait Data {
+    fn lazy_load_was_sent_before(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+        ll_user: &UserId,
+    ) -> Result<bool>;
+
+    fn lazy_load_confirm_delivery(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+        since: u64,
+    ) -> Result<()>;
+
+    fn lazy_load_reset(
+        &self,
+        user_id: &UserId,
+        device_id: &DeviceId,
+        room_id: &RoomId,
+    ) -> Result<()>;
+}

src/service/rooms/lazy_loading/mod.rs

@@ -1,4 +1,13 @@
+mod data;
+pub use data::Data;
 
+use crate::service::*;
+
+pub struct Service<D: Data> {
+    db: D,
+}
+
+impl Service<_> {
     #[tracing::instrument(skip(self))]
     pub fn lazy_load_was_sent_before(
         &self,
@@ -7,14 +16,7 @@
         room_id: &RoomId,
         ll_user: &UserId,
     ) -> Result<bool> {
-        let mut key = user_id.as_bytes().to_vec();
-        key.push(0xff);
-        key.extend_from_slice(device_id.as_bytes());
-        key.push(0xff);
-        key.extend_from_slice(room_id.as_bytes());
-        key.push(0xff);
-        key.extend_from_slice(ll_user.as_bytes());
-        Ok(self.lazyloadedids.get(&key)?.is_some())
+        self.db.lazy_load_was_sent_before(user_id, device_id, room_id, ll_user)
     }
 
     #[tracing::instrument(skip(self))]
@@ -45,27 +47,7 @@
         room_id: &RoomId,
         since: u64,
     ) -> Result<()> {
-        if let Some(user_ids) = self.lazy_load_waiting.lock().unwrap().remove(&(
-            user_id.to_owned(),
-            device_id.to_owned(),
-            room_id.to_owned(),
-            since,
-        )) {
-            let mut prefix = user_id.as_bytes().to_vec();
-            prefix.push(0xff);
-            prefix.extend_from_slice(device_id.as_bytes());
-            prefix.push(0xff);
-            prefix.extend_from_slice(room_id.as_bytes());
-            prefix.push(0xff);
-
-            for ll_id in user_ids {
-                let mut key = prefix.clone();
-                key.extend_from_slice(ll_id.as_bytes());
-                self.lazyloadedids.insert(&key, &[])?;
-            }
-        }
-
-        Ok(())
+        self.db.lazy_load_confirm_delivery(user_d, device_id, room_id, since)
     }
 
     #[tracing::instrument(skip(self))]
@@ -75,17 +57,6 @@
         device_id: &DeviceId,
         room_id: &RoomId,
     ) -> Result<()> {
-        let mut prefix = user_id.as_bytes().to_vec();
-        prefix.push(0xff);
-        prefix.extend_from_slice(device_id.as_bytes());
-        prefix.push(0xff);
-        prefix.extend_from_slice(room_id.as_bytes());
-        prefix.push(0xff);
-
-        for (key, _) in self.lazyloadedids.scan_prefix(prefix) {
-            self.lazyloadedids.remove(&key)?;
-        }
-
-        Ok(())
+        self.db.lazy_load_reset(user_id, device_id, room_id);
     }
-
+}

src/service/rooms/metadata/data.rs

@@ -0,0 +1,3 @@
+pub trait Data {
+    fn exists(&self, room_id: &RoomId) -> Result<bool>;
+}

src/service/rooms/metadata/mod.rs

@@ -1,44 +1,16 @@
+mod data;
+pub use data::Data;
+
+use crate::service::*;
+
+pub struct Service<D: Data> {
+    db: D,
+}
+
+impl Service<_> {
     /// Checks if a room exists.
     #[tracing::instrument(skip(self))]
     pub fn exists(&self, room_id: &RoomId) -> Result<bool> {
-        let prefix = match self.get_shortroomid(room_id)? {
-            Some(b) => b.to_be_bytes().to_vec(),
-            None => return Ok(false),
-        };
-
-        // Look for PDUs in that room.
-        Ok(self
-            .pduid_pdu
-            .iter_from(&prefix, false)
-            .next()
-            .filter(|(k, _)| k.starts_with(&prefix))
-            .is_some())
+        self.db.exists(room_id)
     }
-
-    pub fn get_shortroomid(&self, room_id: &RoomId) -> Result<Option<u64>> {
-        self.roomid_shortroomid
-            .get(room_id.as_bytes())?
-            .map(|bytes| {
-                utils::u64_from_bytes(&bytes)
-                    .map_err(|_| Error::bad_database("Invalid shortroomid in db."))
-            })
-            .transpose()
-    }
-
-    pub fn get_or_create_shortroomid(
-        &self,
-        room_id: &RoomId,
-        globals: &super::globals::Globals,
-    ) -> Result<u64> {
-        Ok(match self.roomid_shortroomid.get(room_id.as_bytes())? {
-            Some(short) => utils::u64_from_bytes(&short)
-                .map_err(|_| Error::bad_database("Invalid shortroomid in db."))?,
-            None => {
-                let short = globals.next_count()?;
-                self.roomid_shortroomid
-                    .insert(room_id.as_bytes(), &short.to_be_bytes())?;
-                short
-            }
-        })
-    }
-
+}

src/service/rooms/outlier/data.rs

@@ -0,0 +1,5 @@
+pub trait Data {
+    fn get_outlier_pdu_json(&self, event_id: &EventId) -> Result<Option<CanonicalJsonObject>>;
+    fn get_outlier_pdu(&self, event_id: &EventId) -> Result<Option<PduEvent>>;
+    fn add_pdu_outlier(&self, event_id: &EventId, pdu: &CanonicalJsonObject) -> Result<()>;
+}

src/service/rooms/outlier/mod.rs

@@ -1,27 +1,26 @@
+mod data;
+pub use data::Data;
+
+use crate::service::*;
+
+pub struct Service<D: Data> {
+    db: D,
+}
+
+impl Service<_> {
     /// Returns the pdu from the outlier tree.
     pub fn get_outlier_pdu_json(&self, event_id: &EventId) -> Result<Option<CanonicalJsonObject>> {
-        self.eventid_outlierpdu
-            .get(event_id.as_bytes())?
-            .map_or(Ok(None), |pdu| {
-                serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
-            })
+        self.db.get_outlier_pdu_json(event_id)
     }
 
     /// Returns the pdu from the outlier tree.
     pub fn get_pdu_outlier(&self, event_id: &EventId) -> Result<Option<PduEvent>> {
-        self.eventid_outlierpdu
-            .get(event_id.as_bytes())?
-            .map_or(Ok(None), |pdu| {
-                serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
-            })
+        self.db.get_outlier_pdu(event_id)
     }
 
     /// Append the PDU as an outlier.
     #[tracing::instrument(skip(self, pdu))]
     pub fn add_pdu_outlier(&self, event_id: &EventId, pdu: &CanonicalJsonObject) -> Result<()> {
-        self.eventid_outlierpdu.insert(
-            event_id.as_bytes(),
-            &serde_json::to_vec(&pdu).expect("CanonicalJsonObject is valid"),
-        )
+        self.db.add_pdu_outlier(event_id, pdu)
     }
-
+}

src/service/rooms/pdu_metadata/data.rs

@@ -0,0 +1,6 @@
+pub trait Data {
+    fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()>;
+    fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> Result<bool>;
+    fn mark_event_soft_failed(&self, event_id: &EventId) -> Result<()>;
+    fn is_event_soft_failed(&self, event_id: &EventId) -> Result<bool>;
+}

src/service/rooms/pdu_metadata/mod.rs

@@ -1,31 +1,30 @@
+mod data;
+pub use data::Data;
 
+use crate::service::*;
+
+pub struct Service<D: Data> {
+    db: D,
+}
+
+impl Service<_> {
     #[tracing::instrument(skip(self, room_id, event_ids))]
     pub fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()> {
-        for prev in event_ids {
-            let mut key = room_id.as_bytes().to_vec();
-            key.extend_from_slice(prev.as_bytes());
-            self.referencedevents.insert(&key, &[])?;
-        }
-
-        Ok(())
+        self.db.mark_as_referenced(room_id, event_ids)
     }
 
     #[tracing::instrument(skip(self))]
     pub fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> Result<bool> {
-        let mut key = room_id.as_bytes().to_vec();
-        key.extend_from_slice(event_id.as_bytes());
-        Ok(self.referencedevents.get(&key)?.is_some())
+        self.db.is_event_referenced(room_id, event_id)
     }
 
     #[tracing::instrument(skip(self))]
     pub fn mark_event_soft_failed(&self, event_id: &EventId) -> Result<()> {
-        self.softfailedeventids.insert(event_id.as_bytes(), &[])
+        self.db.mark_event_soft_failed(event_id)
     }
 
     #[tracing::instrument(skip(self))]
     pub fn is_event_soft_failed(&self, event_id: &EventId) -> Result<bool> {
-        self.softfailedeventids
-            .get(event_id.as_bytes())
-            .map(|o| o.is_some())
+        self.db.is_event_soft_failed(event_id)
     }
-
+}

src/service/rooms/short/mod.rs

@@ -196,3 +196,30 @@
         })
     }
 
+    pub fn get_shortroomid(&self, room_id: &RoomId) -> Result<Option<u64>> {
+        self.roomid_shortroomid
+            .get(room_id.as_bytes())?
+            .map(|bytes| {
+                utils::u64_from_bytes(&bytes)
+                    .map_err(|_| Error::bad_database("Invalid shortroomid in db."))
+            })
+            .transpose()
+    }
+
+    pub fn get_or_create_shortroomid(
+        &self,
+        room_id: &RoomId,
+        globals: &super::globals::Globals,
+    ) -> Result<u64> {
+        Ok(match self.roomid_shortroomid.get(room_id.as_bytes())? {
+            Some(short) => utils::u64_from_bytes(&short)
+                .map_err(|_| Error::bad_database("Invalid shortroomid in db."))?,
+            None => {
+                let short = globals.next_count()?;
+                self.roomid_shortroomid
+                    .insert(room_id.as_bytes(), &short.to_be_bytes())?;
+                short
+            }
+        })
+    }
+