refactor: work on search

12 files changed, 675 insertions, 949 deletions

diff --git a/src/database/key_value/rooms/alias.rs b/src/database/key_value/rooms/alias.rs
new file mode 100644
index 0000000..b00eb3b
--- /dev/null
+++ b/src/database/key_value/rooms/alias.rs
@@ -0,0 +1,66 @@
+impl service::room::alias::Data for KeyValueDatabase {
+    fn set_alias(
+        &self,
+        alias: &RoomAliasId,
+        room_id: Option<&RoomId>
+    ) -> Result<()> {
+        self.alias_roomid
+            .insert(alias.alias().as_bytes(), room_id.as_bytes())?;
+        let mut aliasid = room_id.as_bytes().to_vec();
+        aliasid.push(0xff);
+        aliasid.extend_from_slice(&globals.next_count()?.to_be_bytes());
+        self.aliasid_alias.insert(&aliasid, &*alias.as_bytes())?;
+        Ok(())
+    }
+
+    fn remove_alias(
+        &self,
+        alias: &RoomAliasId,
+    ) -> Result<()> {
+        if let Some(room_id) = self.alias_roomid.get(alias.alias().as_bytes())? {
+            let mut prefix = room_id.to_vec();
+            prefix.push(0xff);
+
+            for (key, _) in self.aliasid_alias.scan_prefix(prefix) {
+                self.aliasid_alias.remove(&key)?;
+            }
+            self.alias_roomid.remove(alias.alias().as_bytes())?;
+        } else {
+            return Err(Error::BadRequest(
+                ErrorKind::NotFound,
+                "Alias does not exist.",
+            ));
+        }
+        Ok(())
+    }
+
+    fn resolve_local_alias(
+        &self,
+        alias: &RoomAliasId
+    ) -> Result<()> {
+        self.alias_roomid
+            .get(alias.alias().as_bytes())?
+            .map(|bytes| {
+                RoomId::parse(utils::string_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Room ID in alias_roomid is invalid unicode.")
+                })?)
+                .map_err(|_| Error::bad_database("Room ID in alias_roomid is invalid."))
+            })
+            .transpose()
+    }
+
+    fn local_aliases_for_room(
+        &self,
+        room_id: &RoomId,
+    ) -> Result<()> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        self.aliasid_alias.scan_prefix(prefix).map(|(_, bytes)| {
+            utils::string_from_bytes(&bytes)
+                .map_err(|_| Error::bad_database("Invalid alias bytes in aliasid_alias."))?
+                .try_into()
+                .map_err(|_| Error::bad_database("Invalid alias in aliasid_alias."))
+        })
+    }
+}
diff --git a/src/database/key_value/rooms/directory.rs b/src/database/key_value/rooms/directory.rs
new file mode 100644
index 0000000..f42de45
--- /dev/null
+++ b/src/database/key_value/rooms/directory.rs
@@ -0,0 +1,24 @@
+impl service::room::directory::Data for KeyValueDatabase {
+    fn set_public(&self, room_id: &RoomId) -> Result<()> {
+        self.publicroomids.insert(room_id.as_bytes(), &[])?;
+    }
+
+    fn set_not_public(&self, room_id: &RoomId) -> Result<()> {
+        self.publicroomids.remove(room_id.as_bytes())?;
+    }
+
+    fn is_public_room(&self, room_id: &RoomId) -> Result<bool> {
+        Ok(self.publicroomids.get(room_id.as_bytes())?.is_some())
+    }
+
+    fn public_rooms(&self) -> impl Iterator<Item = Result<Box<RoomId>>> + '_ {
+        self.publicroomids.iter().map(|(bytes, _)| {
+            RoomId::parse(
+                utils::string_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Room ID in publicroomids is invalid unicode.")
+                })?,
+            )
+            .map_err(|_| Error::bad_database("Room ID in publicroomids is invalid."))
+        })
+    }
+}
diff --git a/src/database/key_value/rooms/edus/presence.rs b/src/database/key_value/rooms/edus/presence.rs
new file mode 100644
index 0000000..61bd9d6
--- /dev/null
+++ b/src/database/key_value/rooms/edus/presence.rs
@@ -0,0 +1,124 @@
+impl service::room::edus::presence::Data for KeyValueDatabase {
+    fn update_presence(
+        &self,
+        user_id: &UserId,
+        room_id: &RoomId,
+        presence: PresenceEvent,
+    ) -> Result<()> {
+        // TODO: Remove old entry? Or maybe just wipe completely from time to time?
+
+        let count = globals.next_count()?.to_be_bytes();
+
+        let mut presence_id = room_id.as_bytes().to_vec();
+        presence_id.push(0xff);
+        presence_id.extend_from_slice(&count);
+        presence_id.push(0xff);
+        presence_id.extend_from_slice(presence.sender.as_bytes());
+
+        self.presenceid_presence.insert(
+            &presence_id,
+            &serde_json::to_vec(&presence).expect("PresenceEvent can be serialized"),
+        )?;
+
+        self.userid_lastpresenceupdate.insert(
+            user_id.as_bytes(),
+            &utils::millis_since_unix_epoch().to_be_bytes(),
+        )?;
+
+        Ok(())
+    }
+
+    fn ping_presence(&self, user_id: &UserId) -> Result<()> {
+        self.userid_lastpresenceupdate.insert(
+            user_id.as_bytes(),
+            &utils::millis_since_unix_epoch().to_be_bytes(),
+        )?;
+
+        Ok(())
+    }
+
+    fn last_presence_update(&self, user_id: &UserId) -> Result<Option<u64>> {
+        self.userid_lastpresenceupdate
+            .get(user_id.as_bytes())?
+            .map(|bytes| {
+                utils::u64_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Invalid timestamp in userid_lastpresenceupdate.")
+                })
+            })
+            .transpose()
+    }
+
+    fn get_presence_event(
+        &self,
+        user_id: &UserId,
+        room_id: &RoomId,
+        count: u64,
+    ) -> Result<Option<PresenceEvent>> {
+        let mut presence_id = room_id.as_bytes().to_vec();
+        presence_id.push(0xff);
+        presence_id.extend_from_slice(&count.to_be_bytes());
+        presence_id.push(0xff);
+        presence_id.extend_from_slice(user_id.as_bytes());
+
+        self.presenceid_presence
+            .get(&presence_id)?
+            .map(|value| parse_presence_event(&value))
+            .transpose()
+    }
+
+    fn presence_since(
+        &self,
+        room_id: &RoomId,
+        since: u64,
+    ) -> Result<HashMap<Box<UserId>, PresenceEvent>> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut first_possible_edu = prefix.clone();
+        first_possible_edu.extend_from_slice(&(since + 1).to_be_bytes()); // +1 so we don't send the event at since
+        let mut hashmap = HashMap::new();
+
+        for (key, value) in self
+            .presenceid_presence
+            .iter_from(&*first_possible_edu, false)
+            .take_while(|(key, _)| key.starts_with(&prefix))
+        {
+            let user_id = UserId::parse(
+                utils::string_from_bytes(
+                    key.rsplit(|&b| b == 0xff)
+                        .next()
+                        .expect("rsplit always returns an element"),
+                )
+                .map_err(|_| Error::bad_database("Invalid UserId bytes in presenceid_presence."))?,
+            )
+            .map_err(|_| Error::bad_database("Invalid UserId in presenceid_presence."))?;
+
+            let presence = parse_presence_event(&value)?;
+
+            hashmap.insert(user_id, presence);
+        }
+
+        Ok(hashmap)
+    }
+}
+
+fn parse_presence_event(bytes: &[u8]) -> Result<PresenceEvent> {
+    let mut presence: PresenceEvent = serde_json::from_slice(bytes)
+        .map_err(|_| Error::bad_database("Invalid presence event in db."))?;
+
+    let current_timestamp: UInt = utils::millis_since_unix_epoch()
+        .try_into()
+        .expect("time is valid");
+
+    if presence.content.presence == PresenceState::Online {
+        // Don't set last_active_ago when the user is online
+        presence.content.last_active_ago = None;
+    } else {
+        // Convert from timestamp to duration
+        presence.content.last_active_ago = presence
+            .content
+            .last_active_ago
+            .map(|timestamp| current_timestamp - timestamp);
+    }
+}
+
diff --git a/src/database/key_value/rooms/edus/read_receipt.rs b/src/database/key_value/rooms/edus/read_receipt.rs
new file mode 100644
index 0000000..556e697
--- /dev/null
+++ b/src/database/key_value/rooms/edus/read_receipt.rs
@@ -0,0 +1,138 @@
+impl service::room::edus::read_receipt::Data for KeyValueDatabase {
+    fn readreceipt_update(
+        &self,
+        user_id: &UserId,
+        room_id: &RoomId,
+        event: ReceiptEvent,
+    ) -> Result<()> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut last_possible_key = prefix.clone();
+        last_possible_key.extend_from_slice(&u64::MAX.to_be_bytes());
+
+        // Remove old entry
+        if let Some((old, _)) = self
+            .readreceiptid_readreceipt
+            .iter_from(&last_possible_key, true)
+            .take_while(|(key, _)| key.starts_with(&prefix))
+            .find(|(key, _)| {
+                key.rsplit(|&b| b == 0xff)
+                    .next()
+                    .expect("rsplit always returns an element")
+                    == user_id.as_bytes()
+            })
+        {
+            // This is the old room_latest
+            self.readreceiptid_readreceipt.remove(&old)?;
+        }
+
+        let mut room_latest_id = prefix;
+        room_latest_id.extend_from_slice(&globals.next_count()?.to_be_bytes());
+        room_latest_id.push(0xff);
+        room_latest_id.extend_from_slice(user_id.as_bytes());
+
+        self.readreceiptid_readreceipt.insert(
+            &room_latest_id,
+            &serde_json::to_vec(&event).expect("EduEvent::to_string always works"),
+        )?;
+
+        Ok(())
+    }
+
+    pub fn readreceipts_since<'a>(
+        &'a self,
+        room_id: &RoomId,
+        since: u64,
+    ) -> impl Iterator<
+        Item=Result<(
+            Box<UserId>,
+            u64,
+            Raw<ruma::events::AnySyncEphemeralRoomEvent>,
+        )>,
+    > + 'a {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+        let prefix2 = prefix.clone();
+
+        let mut first_possible_edu = prefix.clone();
+        first_possible_edu.extend_from_slice(&(since + 1).to_be_bytes()); // +1 so we don't send the event at since
+
+        self.readreceiptid_readreceipt
+            .iter_from(&first_possible_edu, false)
+            .take_while(move |(k, _)| k.starts_with(&prefix2))
+            .map(move |(k, v)| {
+                let count =
+                    utils::u64_from_bytes(&k[prefix.len()..prefix.len() + mem::size_of::<u64>()])
+                        .map_err(|_| Error::bad_database("Invalid readreceiptid count in db."))?;
+                let user_id = UserId::parse(
+                    utils::string_from_bytes(&k[prefix.len() + mem::size_of::<u64>() + 1..])
+                        .map_err(|_| {
+                            Error::bad_database("Invalid readreceiptid userid bytes in db.")
+                        })?,
+                )
+                    .map_err(|_| Error::bad_database("Invalid readreceiptid userid in db."))?;
+
+                let mut json = serde_json::from_slice::<CanonicalJsonObject>(&v).map_err(|_| {
+                    Error::bad_database("Read receipt in roomlatestid_roomlatest is invalid json.")
+                })?;
+                json.remove("room_id");
+
+                Ok((
+                    user_id,
+                    count,
+                    Raw::from_json(
+                        serde_json::value::to_raw_value(&json).expect("json is valid raw value"),
+                    ),
+                ))
+            })
+    }
+
+    fn private_read_set(
+        &self,
+        room_id: &RoomId,
+        user_id: &UserId,
+        count: u64,
+    ) -> Result<()> {
+        let mut key = room_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(user_id.as_bytes());
+
+        self.roomuserid_privateread
+            .insert(&key, &count.to_be_bytes())?;
+
+        self.roomuserid_lastprivatereadupdate
+            .insert(&key, &globals.next_count()?.to_be_bytes())?;
+    }
+
+    fn private_read_get(&self, room_id: &RoomId, user_id: &UserId) -> Result<Option<u64>> {
+        let mut key = room_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(user_id.as_bytes());
+
+        self.roomuserid_privateread
+            .get(&key)?
+            .map_or(Ok(None), |v| {
+                Ok(Some(utils::u64_from_bytes(&v).map_err(|_| {
+                    Error::bad_database("Invalid private read marker bytes")
+                })?))
+            })
+    }
+
+    fn last_privateread_update(&self, user_id: &UserId, room_id: &RoomId) -> Result<u64> {
+        let mut key = room_id.as_bytes().to_vec();
+        key.push(0xff);
+        key.extend_from_slice(user_id.as_bytes());
+
+        Ok(self
+            .roomuserid_lastprivatereadupdate
+            .get(&key)?
+            .map(|bytes| {
+                utils::u64_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Count in roomuserid_lastprivatereadupdate is invalid.")
+                })
+            })
+            .transpose()?
+            .unwrap_or(0))
+    }
+}
diff --git a/src/database/key_value/rooms/edus/typing.rs b/src/database/key_value/rooms/edus/typing.rs
new file mode 100644
index 0000000..8cfb432
--- /dev/null
+++ b/src/database/key_value/rooms/edus/typing.rs
@@ -0,0 +1,94 @@
+impl service::room::edus::typing::Data for KeyValueDatabase {
+    fn typing_add(
+        &self,
+        user_id: &UserId,
+        room_id: &RoomId,
+        timeout: u64,
+        globals: &super::super::globals::Globals,
+    ) -> Result<()> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let count = globals.next_count()?.to_be_bytes();
+
+        let mut room_typing_id = prefix;
+        room_typing_id.extend_from_slice(&timeout.to_be_bytes());
+        room_typing_id.push(0xff);
+        room_typing_id.extend_from_slice(&count);
+
+        self.typingid_userid
+            .insert(&room_typing_id, &*user_id.as_bytes())?;
+
+        self.roomid_lasttypingupdate
+            .insert(room_id.as_bytes(), &count)?;
+
+        Ok(())
+    }
+
+    fn typing_remove(
+        &self,
+        user_id: &UserId,
+        room_id: &RoomId,
+    ) -> Result<()> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let user_id = user_id.to_string();
+
+        let mut found_outdated = false;
+
+        // Maybe there are multiple ones from calling roomtyping_add multiple times
+        for outdated_edu in self
+            .typingid_userid
+            .scan_prefix(prefix)
+            .filter(|(_, v)| &**v == user_id.as_bytes())
+        {
+            self.typingid_userid.remove(&outdated_edu.0)?;
+            found_outdated = true;
+        }
+
+        if found_outdated {
+            self.roomid_lasttypingupdate
+                .insert(room_id.as_bytes(), &globals.next_count()?.to_be_bytes())?;
+        }
+
+        Ok(())
+    }
+
+    fn last_typing_update(
+        &self,
+        room_id: &RoomId,
+    ) -> Result<u64> {
+        Ok(self
+            .roomid_lasttypingupdate
+            .get(room_id.as_bytes())?
+            .map(|bytes| {
+                utils::u64_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Count in roomid_lastroomactiveupdate is invalid.")
+                })
+            })
+            .transpose()?
+            .unwrap_or(0))
+    }
+
+    fn typings_all(
+        &self,
+        room_id: &RoomId,
+    ) -> Result<HashSet<UserId>> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        let mut user_ids = HashSet::new();
+
+        for (_, user_id) in self.typingid_userid.scan_prefix(prefix) {
+            let user_id = UserId::parse(utils::string_from_bytes(&user_id).map_err(|_| {
+                Error::bad_database("User ID in typingid_userid is invalid unicode.")
+            })?)
+                .map_err(|_| Error::bad_database("User ID in typingid_userid is invalid."))?;
+
+            user_ids.insert(user_id);
+        }
+
+        Ok(user_ids)
+    }
+}
diff --git a/src/database/key_value/rooms/lazy_load.rs b/src/database/key_value/rooms/lazy_load.rs
new file mode 100644
index 0000000..8abdce4
--- /dev/null
+++ b/src/database/key_value/rooms/lazy_load.rs
@@ -0,0 +1,68 @@
+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(())
+    }
+}
diff --git a/src/database/key_value/rooms/metadata.rs b/src/database/key_value/rooms/metadata.rs
new file mode 100644
index 0000000..37dd717
--- /dev/null
+++ b/src/database/key_value/rooms/metadata.rs
@@ -0,0 +1,16 @@
+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())
+    }
+}
diff --git a/src/database/key_value/rooms/mod.rs b/src/database/key_value/rooms/mod.rs
new file mode 100644
index 0000000..2a3f81d
--- /dev/null
+++ b/src/database/key_value/rooms/mod.rs
@@ -0,0 +1,17 @@
+mod state;
+mod alias;
+mod directory;
+mod edus;
+mod event_handler;
+mod lazy_loading;
+mod metadata;
+mod outlier;
+mod pdu_metadata;
+mod search;
+mod short;
+mod state;
+mod state_accessor;
+mod state_cache;
+mod state_compressor;
+mod timeline;
+mod user;
diff --git a/src/database/key_value/rooms/outlier.rs b/src/database/key_value/rooms/outlier.rs
new file mode 100644
index 0000000..c979d25
--- /dev/null
+++ b/src/database/key_value/rooms/outlier.rs
@@ -0,0 +1,24 @@
+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"),
+        )
+    }
+}
diff --git a/src/database/key_value/rooms/pdu_metadata.rs b/src/database/key_value/rooms/pdu_metadata.rs
new file mode 100644
index 0000000..6b2171c
--- /dev/null
+++ b/src/database/key_value/rooms/pdu_metadata.rs
@@ -0,0 +1,27 @@
+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())
+    }
+}
diff --git a/src/database/key_value/rooms/search.rs b/src/database/key_value/rooms/search.rs
index 6a32e8b..1ffffe5 100644
--- a/src/database/key_value/rooms/search.rs
+++ b/src/database/key_value/rooms/search.rs
@@ -1,956 +1,23 @@
+impl service::room::search::Data for KeyValueDatabase {
 
-    /// Checks if a room exists.
-    #[tracing::instrument(skip(self))]
-    pub fn first_pdu_in_room(&self, room_id: &RoomId) -> Result<Option<Arc<PduEvent>>> {
-        let prefix = self
-            .get_shortroomid(room_id)?
-            .expect("room exists")
-            .to_be_bytes()
-            .to_vec();
-
-        // Look for PDUs in that room.
-        self.pduid_pdu
-            .iter_from(&prefix, false)
-            .filter(|(k, _)| k.starts_with(&prefix))
-            .map(|(_, pdu)| {
-                serde_json::from_slice(&pdu)
-                    .map_err(|_| Error::bad_database("Invalid first PDU in db."))
-                    .map(Arc::new)
-            })
-            .next()
-            .transpose()
-    }
-
-    #[tracing::instrument(skip(self))]
-    pub fn last_timeline_count(&self, sender_user: &UserId, room_id: &RoomId) -> Result<u64> {
-        match self
-            .lasttimelinecount_cache
-            .lock()
-            .unwrap()
-            .entry(room_id.to_owned())
-        {
-            hash_map::Entry::Vacant(v) => {
-                if let Some(last_count) = self
-                    .pdus_until(&sender_user, &room_id, u64::MAX)?
-                    .filter_map(|r| {
-                        // Filter out buggy events
-                        if r.is_err() {
-                            error!("Bad pdu in pdus_since: {:?}", r);
-                        }
-                        r.ok()
-                    })
-                    .map(|(pduid, _)| self.pdu_count(&pduid))
-                    .next()
-                {
-                    Ok(*v.insert(last_count?))
-                } else {
-                    Ok(0)
-                }
-            }
-            hash_map::Entry::Occupied(o) => Ok(*o.get()),
-        }
-    }
-
-    // TODO Is this the same as the function above?
-    #[tracing::instrument(skip(self))]
-    pub fn latest_pdu_count(&self, room_id: &RoomId) -> Result<u64> {
-        let prefix = self
-            .get_shortroomid(room_id)?
-            .expect("room exists")
-            .to_be_bytes()
-            .to_vec();
-
-        let mut last_possible_key = prefix.clone();
-        last_possible_key.extend_from_slice(&u64::MAX.to_be_bytes());
-
-        self.pduid_pdu
-            .iter_from(&last_possible_key, true)
-            .take_while(move |(k, _)| k.starts_with(&prefix))
-            .next()
-            .map(|b| self.pdu_count(&b.0))
-            .transpose()
-            .map(|op| op.unwrap_or_default())
-    }
-
-
-
-    /// Returns the `count` of this pdu's id.
-    pub fn get_pdu_count(&self, event_id: &EventId) -> Result<Option<u64>> {
-        self.eventid_pduid
-            .get(event_id.as_bytes())?
-            .map(|pdu_id| self.pdu_count(&pdu_id))
-            .transpose()
-    }
-
-    /// Returns the json of a pdu.
-    pub fn get_pdu_json(&self, event_id: &EventId) -> Result<Option<CanonicalJsonObject>> {
-        self.eventid_pduid
-            .get(event_id.as_bytes())?
-            .map_or_else(
-                || self.eventid_outlierpdu.get(event_id.as_bytes()),
-                |pduid| {
-                    Ok(Some(self.pduid_pdu.get(&pduid)?.ok_or_else(|| {
-                        Error::bad_database("Invalid pduid in eventid_pduid.")
-                    })?))
-                },
-            )?
-            .map(|pdu| {
-                serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
-            })
-            .transpose()
-    }
-
-    /// Returns the json of a pdu.
-    pub fn get_non_outlier_pdu_json(
-        &self,
-        event_id: &EventId,
-    ) -> Result<Option<CanonicalJsonObject>> {
-        self.eventid_pduid
-            .get(event_id.as_bytes())?
-            .map(|pduid| {
-                self.pduid_pdu
-                    .get(&pduid)?
-                    .ok_or_else(|| Error::bad_database("Invalid pduid in eventid_pduid."))
-            })
-            .transpose()?
-            .map(|pdu| {
-                serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
-            })
-            .transpose()
-    }
-
-    /// Returns the pdu's id.
-    pub fn get_pdu_id(&self, event_id: &EventId) -> Result<Option<Vec<u8>>> {
-        self.eventid_pduid.get(event_id.as_bytes())
-    }
-
-    /// Returns the pdu.
-    ///
-    /// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
-    pub fn get_non_outlier_pdu(&self, event_id: &EventId) -> Result<Option<PduEvent>> {
-        self.eventid_pduid
-            .get(event_id.as_bytes())?
-            .map(|pduid| {
-                self.pduid_pdu
-                    .get(&pduid)?
-                    .ok_or_else(|| Error::bad_database("Invalid pduid in eventid_pduid."))
-            })
-            .transpose()?
-            .map(|pdu| {
-                serde_json::from_slice(&pdu).map_err(|_| Error::bad_database("Invalid PDU in db."))
-            })
-            .transpose()
-    }
-
-    /// Returns the pdu.
-    ///
-    /// Checks the `eventid_outlierpdu` Tree if not found in the timeline.
-    pub fn get_pdu(&self, event_id: &EventId) -> Result<Option<Arc<PduEvent>>> {
-        if let Some(p) = self.pdu_cache.lock().unwrap().get_mut(event_id) {
-            return Ok(Some(Arc::clone(p)));
-        }
-
-        if let Some(pdu) = self
-            .eventid_pduid
-            .get(event_id.as_bytes())?
-            .map_or_else(
-                || self.eventid_outlierpdu.get(event_id.as_bytes()),
-                |pduid| {
-                    Ok(Some(self.pduid_pdu.get(&pduid)?.ok_or_else(|| {
-                        Error::bad_database("Invalid pduid in eventid_pduid.")
-                    })?))
-                },
-            )?
-            .map(|pdu| {
-                serde_json::from_slice(&pdu)
-                    .map_err(|_| Error::bad_database("Invalid PDU in db."))
-                    .map(Arc::new)
-            })
-            .transpose()?
-        {
-            self.pdu_cache
-                .lock()
-                .unwrap()
-                .insert(event_id.to_owned(), Arc::clone(&pdu));
-            Ok(Some(pdu))
-        } else {
-            Ok(None)
-        }
-    }
-
-    /// Returns the pdu.
-    ///
-    /// This does __NOT__ check the outliers `Tree`.
-    pub fn get_pdu_from_id(&self, pdu_id: &[u8]) -> Result<Option<PduEvent>> {
-        self.pduid_pdu.get(pdu_id)?.map_or(Ok(None), |pdu| {
-            Ok(Some(
-                serde_json::from_slice(&pdu)
-                    .map_err(|_| Error::bad_database("Invalid PDU in db."))?,
-            ))
-        })
-    }
-
-    /// Returns the pdu as a `BTreeMap<String, CanonicalJsonValue>`.
-    pub fn get_pdu_json_from_id(&self, pdu_id: &[u8]) -> Result<Option<CanonicalJsonObject>> {
-        self.pduid_pdu.get(pdu_id)?.map_or(Ok(None), |pdu| {
-            Ok(Some(
-                serde_json::from_slice(&pdu)
-                    .map_err(|_| Error::bad_database("Invalid PDU in db."))?,
-            ))
-        })
-    }
-
-    /// Returns the `count` of this pdu's id.
-    pub fn pdu_count(&self, pdu_id: &[u8]) -> Result<u64> {
-        utils::u64_from_bytes(&pdu_id[pdu_id.len() - size_of::<u64>()..])
-            .map_err(|_| Error::bad_database("PDU has invalid count bytes."))
-    }
-
-    /// Removes a pdu and creates a new one with the same id.
-    #[tracing::instrument(skip(self))]
-    fn replace_pdu(&self, pdu_id: &[u8], pdu: &PduEvent) -> Result<()> {
-        if self.pduid_pdu.get(pdu_id)?.is_some() {
-            self.pduid_pdu.insert(
-                pdu_id,
-                &serde_json::to_vec(pdu).expect("PduEvent::to_vec always works"),
-            )?;
-            Ok(())
-        } else {
-            Err(Error::BadRequest(
-                ErrorKind::NotFound,
-                "PDU does not exist.",
-            ))
-        }
-    }
-
-    /// Creates a new persisted data unit and adds it to a room.
-    ///
-    /// By this point the incoming event should be fully authenticated, no auth happens
-    /// in `append_pdu`.
-    ///
-    /// Returns pdu id
-    #[tracing::instrument(skip(self, pdu, pdu_json, leaves, db))]
-    pub fn append_pdu<'a>(
-        &self,
-        pdu: &PduEvent,
-        mut pdu_json: CanonicalJsonObject,
-        leaves: impl IntoIterator<Item = &'a EventId> + Debug,
-        db: &Database,
-    ) -> Result<Vec<u8>> {
-        let shortroomid = self.get_shortroomid(&pdu.room_id)?.expect("room exists");
-
-        // Make unsigned fields correct. This is not properly documented in the spec, but state
-        // events need to have previous content in the unsigned field, so clients can easily
-        // interpret things like membership changes
-        if let Some(state_key) = &pdu.state_key {
-            if let CanonicalJsonValue::Object(unsigned) = pdu_json
-                .entry("unsigned".to_owned())
-                .or_insert_with(|| CanonicalJsonValue::Object(Default::default()))
-            {
-                if let Some(shortstatehash) = self.pdu_shortstatehash(&pdu.event_id).unwrap() {
-                    if let Some(prev_state) = self
-                        .state_get(shortstatehash, &pdu.kind.to_string().into(), state_key)
-                        .unwrap()
-                    {
-                        unsigned.insert(
-                            "prev_content".to_owned(),
-                            CanonicalJsonValue::Object(
-                                utils::to_canonical_object(prev_state.content.clone())
-                                    .expect("event is valid, we just created it"),
-                            ),
-                        );
-                    }
-                }
-            } else {
-                error!("Invalid unsigned type in pdu.");
-            }
-        }
-
-        // We must keep track of all events that have been referenced.
-        self.mark_as_referenced(&pdu.room_id, &pdu.prev_events)?;
-        self.replace_pdu_leaves(&pdu.room_id, leaves)?;
-
-        let mutex_insert = Arc::clone(
-            db.globals
-                .roomid_mutex_insert
-                .write()
-                .unwrap()
-                .entry(pdu.room_id.clone())
-                .or_default(),
-        );
-        let insert_lock = mutex_insert.lock().unwrap();
-
-        let count1 = db.globals.next_count()?;
-        // Mark as read first so the sending client doesn't get a notification even if appending
-        // fails
-        self.edus
-            .private_read_set(&pdu.room_id, &pdu.sender, count1, &db.globals)?;
-        self.reset_notification_counts(&pdu.sender, &pdu.room_id)?;
-
-        let count2 = db.globals.next_count()?;
-        let mut pdu_id = shortroomid.to_be_bytes().to_vec();
-        pdu_id.extend_from_slice(&count2.to_be_bytes());
-
-        // There's a brief moment of time here where the count is updated but the pdu does not
-        // exist. This could theoretically lead to dropped pdus, but it's extremely rare
-        //
-        // Update: We fixed this using insert_lock
-
-        self.pduid_pdu.insert(
-            &pdu_id,
-            &serde_json::to_vec(&pdu_json).expect("CanonicalJsonObject is always a valid"),
-        )?;
-        self.lasttimelinecount_cache
-            .lock()
-            .unwrap()
-            .insert(pdu.room_id.clone(), count2);
-
-        self.eventid_pduid
-            .insert(pdu.event_id.as_bytes(), &pdu_id)?;
-        self.eventid_outlierpdu.remove(pdu.event_id.as_bytes())?;
-
-        drop(insert_lock);
-
-        // See if the event matches any known pushers
-        let power_levels: RoomPowerLevelsEventContent = db
-            .rooms
-            .room_state_get(&pdu.room_id, &StateEventType::RoomPowerLevels, "")?
-            .map(|ev| {
-                serde_json::from_str(ev.content.get())
-                    .map_err(|_| Error::bad_database("invalid m.room.power_levels event"))
-            })
-            .transpose()?
-            .unwrap_or_default();
-
-        let sync_pdu = pdu.to_sync_room_event();
-
-        let mut notifies = Vec::new();
-        let mut highlights = Vec::new();
-
-        for user in self.get_our_real_users(&pdu.room_id, db)?.iter() {
-            // Don't notify the user of their own events
-            if user == &pdu.sender {
-                continue;
-            }
-
-            let rules_for_user = db
-                .account_data
-                .get(
-                    None,
-                    user,
-                    GlobalAccountDataEventType::PushRules.to_string().into(),
-                )?
-                .map(|ev: PushRulesEvent| ev.content.global)
-                .unwrap_or_else(|| Ruleset::server_default(user));
-
-            let mut highlight = false;
-            let mut notify = false;
-
-            for action in pusher::get_actions(
-                user,
-                &rules_for_user,
-                &power_levels,
-                &sync_pdu,
-                &pdu.room_id,
-                db,
-            )? {
-                match action {
-                    Action::DontNotify => notify = false,
-                    // TODO: Implement proper support for coalesce
-                    Action::Notify | Action::Coalesce => notify = true,
-                    Action::SetTweak(Tweak::Highlight(true)) => {
-                        highlight = true;
-                    }
-                    _ => {}
-                };
-            }
-
-            let mut userroom_id = user.as_bytes().to_vec();
-            userroom_id.push(0xff);
-            userroom_id.extend_from_slice(pdu.room_id.as_bytes());
-
-            if notify {
-                notifies.push(userroom_id.clone());
-            }
-
-            if highlight {
-                highlights.push(userroom_id);
-            }
-
-            for senderkey in db.pusher.get_pusher_senderkeys(user) {
-                db.sending.send_push_pdu(&*pdu_id, senderkey)?;
-            }
-        }
-
-        self.userroomid_notificationcount
-            .increment_batch(&mut notifies.into_iter())?;
-        self.userroomid_highlightcount
-            .increment_batch(&mut highlights.into_iter())?;
-
-        match pdu.kind {
-            RoomEventType::RoomRedaction => {
-                if let Some(redact_id) = &pdu.redacts {
-                    self.redact_pdu(redact_id, pdu)?;
-                }
-            }
-            RoomEventType::RoomMember => {
-                if let Some(state_key) = &pdu.state_key {
-                    #[derive(Deserialize)]
-                    struct ExtractMembership {
-                        membership: MembershipState,
-                    }
-
-                    // if the state_key fails
-                    let target_user_id = UserId::parse(state_key.clone())
-                        .expect("This state_key was previously validated");
-
-                    let content = serde_json::from_str::<ExtractMembership>(pdu.content.get())
-                        .map_err(|_| Error::bad_database("Invalid content in pdu."))?;
-
-                    let invite_state = match content.membership {
-                        MembershipState::Invite => {
-                            let state = self.calculate_invite_state(pdu)?;
-                            Some(state)
-                        }
-                        _ => None,
-                    };
-
-                    // Update our membership info, we do this here incase a user is invited
-                    // and immediately leaves we need the DB to record the invite event for auth
-                    self.update_membership(
-                        &pdu.room_id,
-                        &target_user_id,
-                        content.membership,
-                        &pdu.sender,
-                        invite_state,
-                        db,
-                        true,
-                    )?;
-                }
-            }
-            RoomEventType::RoomMessage => {
-                #[derive(Deserialize)]
-                struct ExtractBody<'a> {
-                    #[serde(borrow)]
-                    body: Option<Cow<'a, str>>,
-                }
-
-                let content = serde_json::from_str::<ExtractBody<'_>>(pdu.content.get())
-                    .map_err(|_| Error::bad_database("Invalid content in pdu."))?;
-
-                if let Some(body) = content.body {
-                    let mut batch = body
-                        .split_terminator(|c: char| !c.is_alphanumeric())
-                        .filter(|s| !s.is_empty())
-                        .filter(|word| word.len() <= 50)
-                        .map(str::to_lowercase)
-                        .map(|word| {
-                            let mut key = shortroomid.to_be_bytes().to_vec();
-                            key.extend_from_slice(word.as_bytes());
-                            key.push(0xff);
-                            key.extend_from_slice(&pdu_id);
-                            (key, Vec::new())
-                        });
-
-                    self.tokenids.insert_batch(&mut batch)?;
-
-                    let admin_room = self.id_from_alias(
-                        <&RoomAliasId>::try_from(
-                            format!("#admins:{}", db.globals.server_name()).as_str(),
-                        )
-                        .expect("#admins:server_name is a valid room alias"),
-                    )?;
-                    let server_user = format!("@conduit:{}", db.globals.server_name());
-
-                    let to_conduit = body.starts_with(&format!("{}: ", server_user));
-
-                    // This will evaluate to false if the emergency password is set up so that
-                    // the administrator can execute commands as conduit
-                    let from_conduit =
-                        pdu.sender == server_user && db.globals.emergency_password().is_none();
-
-                    if to_conduit && !from_conduit && admin_room.as_ref() == Some(&pdu.room_id) {
-                        db.admin.process_message(body.to_string());
-                    }
-                }
-            }
-            _ => {}
-        }
-
-        for appservice in db.appservice.all()? {
-            if self.appservice_in_room(room_id, &appservice, db)? {
-                db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
-                continue;
-            }
-
-            // If the RoomMember event has a non-empty state_key, it is targeted at someone.
-            // If it is our appservice user, we send this PDU to it.
-            if pdu.kind == RoomEventType::RoomMember {
-                if let Some(state_key_uid) = &pdu
-                    .state_key
-                    .as_ref()
-                    .and_then(|state_key| UserId::parse(state_key.as_str()).ok())
-                {
-                    if let Some(appservice_uid) = appservice
-                        .1
-                        .get("sender_localpart")
-                        .and_then(|string| string.as_str())
-                        .and_then(|string| {
-                            UserId::parse_with_server_name(string, db.globals.server_name()).ok()
-                        })
-                    {
-                        if state_key_uid == &appservice_uid {
-                            db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
-                            continue;
-                        }
-                    }
-                }
-            }
-
-            if let Some(namespaces) = appservice.1.get("namespaces") {
-                let users = namespaces
-                    .get("users")
-                    .and_then(|users| users.as_sequence())
-                    .map_or_else(Vec::new, |users| {
-                        users
-                            .iter()
-                            .filter_map(|users| Regex::new(users.get("regex")?.as_str()?).ok())
-                            .collect::<Vec<_>>()
-                    });
-                let aliases = namespaces
-                    .get("aliases")
-                    .and_then(|aliases| aliases.as_sequence())
-                    .map_or_else(Vec::new, |aliases| {
-                        aliases
-                            .iter()
-                            .filter_map(|aliases| Regex::new(aliases.get("regex")?.as_str()?).ok())
-                            .collect::<Vec<_>>()
-                    });
-                let rooms = namespaces
-                    .get("rooms")
-                    .and_then(|rooms| rooms.as_sequence());
-
-                let matching_users = |users: &Regex| {
-                    users.is_match(pdu.sender.as_str())
-                        || pdu.kind == RoomEventType::RoomMember
-                            && pdu
-                                .state_key
-                                .as_ref()
-                                .map_or(false, |state_key| users.is_match(state_key))
-                };
-                let matching_aliases = |aliases: &Regex| {
-                    self.room_aliases(room_id)
-                        .filter_map(|r| r.ok())
-                        .any(|room_alias| aliases.is_match(room_alias.as_str()))
-                };
-
-                if aliases.iter().any(matching_aliases)
-                    || rooms.map_or(false, |rooms| rooms.contains(&room_id.as_str().into()))
-                    || users.iter().any(matching_users)
-                {
-                    db.sending.send_pdu_appservice(&appservice.0, &pdu_id)?;
-                }
-            }
-        }
-
-
-        Ok(pdu_id)
-    }
-
-    pub fn create_hash_and_sign_event(
-            &self,
-            pdu_builder: PduBuilder,
-            sender: &UserId,
-            room_id: &RoomId,
-            db: &Database,
-            _mutex_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
-    ) -> (PduEvent, CanonicalJsonObj) {
-        let PduBuilder {
-            event_type,
-            content,
-            unsigned,
-            state_key,
-            redacts,
-        } = pdu_builder;
-
-        let prev_events: Vec<_> = db
-            .rooms
-            .get_pdu_leaves(room_id)?
-            .into_iter()
-            .take(20)
-            .collect();
-
-        let create_event = db
-            .rooms
-            .room_state_get(room_id, &StateEventType::RoomCreate, "")?;
-
-        let create_event_content: Option<RoomCreateEventContent> = create_event
-            .as_ref()
-            .map(|create_event| {
-                serde_json::from_str(create_event.content.get()).map_err(|e| {
-                    warn!("Invalid create event: {}", e);
-                    Error::bad_database("Invalid create event in db.")
-                })
-            })
-            .transpose()?;
-
-        // If there was no create event yet, assume we are creating a room with the default
-        // version right now
-        let room_version_id = create_event_content
-            .map_or(db.globals.default_room_version(), |create_event| {
-                create_event.room_version
+    fn index_pdu<'a>(&self, room_id: &RoomId, pdu_id: u64, message_body: String) -> Result<()> {
+        let mut batch = body
+            .split_terminator(|c: char| !c.is_alphanumeric())
+            .filter(|s| !s.is_empty())
+            .filter(|word| word.len() <= 50)
+            .map(str::to_lowercase)
+            .map(|word| {
+                let mut key = shortroomid.to_be_bytes().to_vec();
+                key.extend_from_slice(word.as_bytes());
+                key.push(0xff);
+                key.extend_from_slice(&pdu_id);
+                (key, Vec::new())
             });
-        let room_version =
-            RoomVersion::new(&room_version_id).expect("room version is supported");
-
-        let auth_events =
-            self.get_auth_events(room_id, &event_type, sender, state_key.as_deref(), &content)?;
-
-        // Our depth is the maximum depth of prev_events + 1
-        let depth = prev_events
-            .iter()
-            .filter_map(|event_id| Some(db.rooms.get_pdu(event_id).ok()??.depth))
-            .max()
-            .unwrap_or_else(|| uint!(0))
-            + uint!(1);
-
-        let mut unsigned = unsigned.unwrap_or_default();
 
-        if let Some(state_key) = &state_key {
-            if let Some(prev_pdu) =
-                self.room_state_get(room_id, &event_type.to_string().into(), state_key)?
-            {
-                unsigned.insert(
-                    "prev_content".to_owned(),
-                    serde_json::from_str(prev_pdu.content.get()).expect("string is valid json"),
-                );
-                unsigned.insert(
-                    "prev_sender".to_owned(),
-                    serde_json::to_value(&prev_pdu.sender).expect("UserId::to_value always works"),
-                );
-            }
-        }
-
-        let pdu = PduEvent {
-            event_id: ruma::event_id!("$thiswillbefilledinlater").into(),
-            room_id: room_id.to_owned(),
-            sender: sender_user.to_owned(),
-            origin_server_ts: utils::millis_since_unix_epoch()
-                .try_into()
-                .expect("time is valid"),
-            kind: event_type,
-            content,
-            state_key,
-            prev_events,
-            depth,
-            auth_events: auth_events
-                .iter()
-                .map(|(_, pdu)| pdu.event_id.clone())
-                .collect(),
-            redacts,
-            unsigned: if unsigned.is_empty() {
-                None
-            } else {
-                Some(to_raw_value(&unsigned).expect("to_raw_value always works"))
-            },
-            hashes: EventHash {
-                sha256: "aaa".to_owned(),
-            },
-            signatures: None,
-        };
-
-        let auth_check = state_res::auth_check(
-            &room_version,
-            &pdu,
-            None::<PduEvent>, // TODO: third_party_invite
-            |k, s| auth_events.get(&(k.clone(), s.to_owned())),
-        )
-        .map_err(|e| {
-            error!("{:?}", e);
-            Error::bad_database("Auth check failed.")
-        })?;
-
-        if !auth_check {
-            return Err(Error::BadRequest(
-                ErrorKind::Forbidden,
-                "Event is not authorized.",
-            ));
-        }
-
-        // Hash and sign
-        let mut pdu_json =
-            utils::to_canonical_object(&pdu).expect("event is valid, we just created it");
-
-        pdu_json.remove("event_id");
-
-        // Add origin because synapse likes that (and it's required in the spec)
-        pdu_json.insert(
-            "origin".to_owned(),
-            to_canonical_value(db.globals.server_name())
-                .expect("server name is a valid CanonicalJsonValue"),
-        );
-
-        match ruma::signatures::hash_and_sign_event(
-            db.globals.server_name().as_str(),
-            db.globals.keypair(),
-            &mut pdu_json,
-            &room_version_id,
-        ) {
-            Ok(_) => {}
-            Err(e) => {
-                return match e {
-                    ruma::signatures::Error::PduSize => Err(Error::BadRequest(
-                        ErrorKind::TooLarge,
-                        "Message is too long",
-                    )),
-                    _ => Err(Error::BadRequest(
-                        ErrorKind::Unknown,
-                        "Signing event failed",
-                    )),
-                }
-            }
-        }
-
-        // Generate event id
-        pdu.event_id = EventId::parse_arc(format!(
-            "${}",
-            ruma::signatures::reference_hash(&pdu_json, &room_version_id)
-                .expect("ruma can calculate reference hashes")
-        ))
-        .expect("ruma's reference hashes are valid event ids");
-
-        pdu_json.insert(
-            "event_id".to_owned(),
-            CanonicalJsonValue::String(pdu.event_id.as_str().to_owned()),
-        );
-
-        // Generate short event id
-        let _shorteventid = self.get_or_create_shorteventid(&pdu.event_id, &db.globals)?;
-    }
-
-    /// Creates a new persisted data unit and adds it to a room. This function takes a
-    /// roomid_mutex_state, meaning that only this function is able to mutate the room state.
-    #[tracing::instrument(skip(self, db, _mutex_lock))]
-    pub fn build_and_append_pdu(
-        &self,
-        pdu_builder: PduBuilder,
-        sender: &UserId,
-        room_id: &RoomId,
-        db: &Database,
-        _mutex_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
-    ) -> Result<Arc<EventId>> {
-
-        let (pdu, pdu_json) = create_hash_and_sign_event()?;
-
-
-        // We append to state before appending the pdu, so we don't have a moment in time with the
-        // pdu without it's state. This is okay because append_pdu can't fail.
-        let statehashid = self.append_to_state(&pdu, &db.globals)?;
-
-        let pdu_id = self.append_pdu(
-            &pdu,
-            pdu_json,
-            // Since this PDU references all pdu_leaves we can update the leaves
-            // of the room
-            iter::once(&*pdu.event_id),
-            db,
-        )?;
-
-        // We set the room state after inserting the pdu, so that we never have a moment in time
-        // where events in the current room state do not exist
-        self.set_room_state(room_id, statehashid)?;
-
-        let mut servers: HashSet<Box<ServerName>> =
-            self.room_servers(room_id).filter_map(|r| r.ok()).collect();
-
-        // In case we are kicking or banning a user, we need to inform their server of the change
-        if pdu.kind == RoomEventType::RoomMember {
-            if let Some(state_key_uid) = &pdu
-                .state_key
-                .as_ref()
-                .and_then(|state_key| UserId::parse(state_key.as_str()).ok())
-            {
-                servers.insert(Box::from(state_key_uid.server_name()));
-            }
-        }
-
-        // Remove our server from the server list since it will be added to it by room_servers() and/or the if statement above
-        servers.remove(db.globals.server_name());
-
-        db.sending.send_pdu(servers.into_iter(), &pdu_id)?;
-
-        Ok(pdu.event_id)
-    }
-
-    /// Append the incoming event setting the state snapshot to the state from the
-    /// server that sent the event.
-    #[tracing::instrument(skip_all)]
-    fn append_incoming_pdu<'a>(
-        db: &Database,
-        pdu: &PduEvent,
-        pdu_json: CanonicalJsonObject,
-        new_room_leaves: impl IntoIterator<Item = &'a EventId> + Clone + Debug,
-        state_ids_compressed: HashSet<CompressedStateEvent>,
-        soft_fail: bool,
-        _mutex_lock: &MutexGuard<'_, ()>, // Take mutex guard to make sure users get the room state mutex
-    ) -> Result<Option<Vec<u8>>> {
-        // We append to state before appending the pdu, so we don't have a moment in time with the
-        // pdu without it's state. This is okay because append_pdu can't fail.
-        db.rooms.set_event_state(
-            &pdu.event_id,
-            &pdu.room_id,
-            state_ids_compressed,
-            &db.globals,
-        )?;
-
-        if soft_fail {
-            db.rooms
-                .mark_as_referenced(&pdu.room_id, &pdu.prev_events)?;
-            db.rooms.replace_pdu_leaves(&pdu.room_id, new_room_leaves)?;
-            return Ok(None);
-        }
-
-        let pdu_id = db.rooms.append_pdu(pdu, pdu_json, new_room_leaves, db)?;
-
-        Ok(Some(pdu_id))
-    }
-
-    /// Returns an iterator over all PDUs in a room.
-    #[tracing::instrument(skip(self))]
-    pub fn all_pdus<'a>(
-        &'a self,
-        user_id: &UserId,
-        room_id: &RoomId,
-    ) -> Result<impl Iterator<Item = Result<(Vec<u8>, PduEvent)>> + 'a> {
-        self.pdus_since(user_id, room_id, 0)
+        self.tokenids.insert_batch(&mut batch)?;
     }
 
-    /// Returns an iterator over all events in a room that happened after the event with id `since`
-    /// in chronological order.
-    #[tracing::instrument(skip(self))]
-    pub fn pdus_since<'a>(
-        &'a self,
-        user_id: &UserId,
-        room_id: &RoomId,
-        since: u64,
-    ) -> Result<impl Iterator<Item = Result<(Vec<u8>, PduEvent)>> + 'a> {
-        let prefix = self
-            .get_shortroomid(room_id)?
-            .expect("room exists")
-            .to_be_bytes()
-            .to_vec();
-
-        // Skip the first pdu if it's exactly at since, because we sent that last time
-        let mut first_pdu_id = prefix.clone();
-        first_pdu_id.extend_from_slice(&(since + 1).to_be_bytes());
-
-        let user_id = user_id.to_owned();
-
-        Ok(self
-            .pduid_pdu
-            .iter_from(&first_pdu_id, false)
-            .take_while(move |(k, _)| k.starts_with(&prefix))
-            .map(move |(pdu_id, v)| {
-                let mut pdu = serde_json::from_slice::<PduEvent>(&v)
-                    .map_err(|_| Error::bad_database("PDU in db is invalid."))?;
-                if pdu.sender != user_id {
-                    pdu.remove_transaction_id()?;
-                }
-                Ok((pdu_id, pdu))
-            }))
-    }
-
-    /// Returns an iterator over all events and their tokens in a room that happened before the
-    /// event with id `until` in reverse-chronological order.
-    #[tracing::instrument(skip(self))]
-    pub fn pdus_until<'a>(
-        &'a self,
-        user_id: &UserId,
-        room_id: &RoomId,
-        until: u64,
-    ) -> Result<impl Iterator<Item = Result<(Vec<u8>, PduEvent)>> + 'a> {
-        // Create the first part of the full pdu id
-        let prefix = self
-            .get_shortroomid(room_id)?
-            .expect("room exists")
-            .to_be_bytes()
-            .to_vec();
-
-        let mut current = prefix.clone();
-        current.extend_from_slice(&(until.saturating_sub(1)).to_be_bytes()); // -1 because we don't want event at `until`
-
-        let current: &[u8] = &current;
-
-        let user_id = user_id.to_owned();
-
-        Ok(self
-            .pduid_pdu
-            .iter_from(current, true)
-            .take_while(move |(k, _)| k.starts_with(&prefix))
-            .map(move |(pdu_id, v)| {
-                let mut pdu = serde_json::from_slice::<PduEvent>(&v)
-                    .map_err(|_| Error::bad_database("PDU in db is invalid."))?;
-                if pdu.sender != user_id {
-                    pdu.remove_transaction_id()?;
-                }
-                Ok((pdu_id, pdu))
-            }))
-    }
-
-    /// Returns an iterator over all events and their token in a room that happened after the event
-    /// with id `from` in chronological order.
-    #[tracing::instrument(skip(self))]
-    pub fn pdus_after<'a>(
-        &'a self,
-        user_id: &UserId,
-        room_id: &RoomId,
-        from: u64,
-    ) -> Result<impl Iterator<Item = Result<(Vec<u8>, PduEvent)>> + 'a> {
-        // Create the first part of the full pdu id
-        let prefix = self
-            .get_shortroomid(room_id)?
-            .expect("room exists")
-            .to_be_bytes()
-            .to_vec();
-
-        let mut current = prefix.clone();
-        current.extend_from_slice(&(from + 1).to_be_bytes()); // +1 so we don't send the base event
-
-        let current: &[u8] = &current;
-
-        let user_id = user_id.to_owned();
-
-        Ok(self
-            .pduid_pdu
-            .iter_from(current, false)
-            .take_while(move |(k, _)| k.starts_with(&prefix))
-            .map(move |(pdu_id, v)| {
-                let mut pdu = serde_json::from_slice::<PduEvent>(&v)
-                    .map_err(|_| Error::bad_database("PDU in db is invalid."))?;
-                if pdu.sender != user_id {
-                    pdu.remove_transaction_id()?;
-                }
-                Ok((pdu_id, pdu))
-            }))
-    }
-
-    /// Replace a PDU with the redacted form.
-    #[tracing::instrument(skip(self, reason))]
-    pub fn redact_pdu(&self, event_id: &EventId, reason: &PduEvent) -> Result<()> {
-        if let Some(pdu_id) = self.get_pdu_id(event_id)? {
-            let mut pdu = self
-                .get_pdu_from_id(&pdu_id)?
-                .ok_or_else(|| Error::bad_database("PDU ID points to invalid PDU."))?;
-            pdu.redact(reason)?;
-            self.replace_pdu(&pdu_id, &pdu)?;
-        }
-        // If event does not exist, just noop
-        Ok(())
-    }
-
-
-    #[tracing::instrument(skip(self))]
-    pub fn search_pdus<'a>(
+    fn search_pdus<'a>(
         &'a self,
         room_id: &RoomId,
         search_string: &str,
@@ -997,4 +64,3 @@
             )
         }))
     }
-
diff --git a/src/database/key_value/rooms/state.rs b/src/database/key_value/rooms/state.rs
new file mode 100644
index 0000000..5daf6c6
--- /dev/null
+++ b/src/database/key_value/rooms/state.rs
@@ -0,0 +1,62 @@
+impl service::room::state::Data for KeyValueDatabase {
+    fn get_room_shortstatehash(&self, room_id: &RoomId) -> Result<Option<u64>> {
+        self.roomid_shortstatehash
+            .get(room_id.as_bytes())?
+            .map_or(Ok(None), |bytes| {
+                Ok(Some(utils::u64_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("Invalid shortstatehash in roomid_shortstatehash")
+                })?))
+            })
+    }
+
+    fn set_room_state(&self, room_id: &RoomId, new_shortstatehash: u64
+        _mutex_lock: &MutexGuard<'_, StateLock>, // Take mutex guard to make sure users get the room state mutex
+        ) -> Result<()> {
+        self.roomid_shortstatehash
+            .insert(room_id.as_bytes(), &new_shortstatehash.to_be_bytes())?;
+        Ok(())
+    }
+
+    fn set_event_state(&self) -> Result<()> {
+        db.shorteventid_shortstatehash
+            .insert(&shorteventid.to_be_bytes(), &shortstatehash.to_be_bytes())?;
+        Ok(())
+    }
+
+    fn get_pdu_leaves(&self, room_id: &RoomId) -> Result<HashSet<Arc<EventId>>> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        self.roomid_pduleaves
+            .scan_prefix(prefix)
+            .map(|(_, bytes)| {
+                EventId::parse_arc(utils::string_from_bytes(&bytes).map_err(|_| {
+                    Error::bad_database("EventID in roomid_pduleaves is invalid unicode.")
+                })?)
+                .map_err(|_| Error::bad_database("EventId in roomid_pduleaves is invalid."))
+            })
+            .collect()
+    }
+
+    fn set_forward_extremities(
+        &self,
+        room_id: &RoomId,
+        event_ids: impl IntoIterator<Item = &'a EventId> + Debug,
+        _mutex_lock: &MutexGuard<'_, StateLock>, // Take mutex guard to make sure users get the room state mutex
+    ) -> Result<()> {
+        let mut prefix = room_id.as_bytes().to_vec();
+        prefix.push(0xff);
+
+        for (key, _) in self.roomid_pduleaves.scan_prefix(prefix.clone()) {
+            self.roomid_pduleaves.remove(&key)?;
+        }
+
+        for event_id in event_ids {
+            let mut key = prefix.to_owned();
+            key.extend_from_slice(event_id.as_bytes());
+            self.roomid_pduleaves.insert(&key, event_id.as_bytes())?;
+        }
+
+        Ok(())
+    }
+}