cotugno/conduit
2
Fork 0
mirror of https://gitlab.com/famedly/conduit.git synced 2024-11-17 02:38:18 -07:00
Code Issues Packages Projects Releases Wiki Activity

Work on rooms/state, database, alias, directory, edus services, event_handler, lazy_loading, metadata, outlier, and pdu_metadata

Browse Source
This commit is contained in:
Timo Kösters 2022-06-25 16:12:23 +02:00 committed by Nyaaori
parent 604b1a5cf1
commit 865e35df17
No known key found for this signature in database
GPG Key ID: E7819C3ED4D1F82E
22 changed files with 1544 additions and 2282 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/database/abstraction.rs
View File

@ -26,7 +26,7 @@ pub mod persy;
))]
pub mod watchers;
pub trait DatabaseEngine: Send + Sync {
pub trait KeyValueDatabaseEngine: Send + Sync {
fn open(config: &Config) -> Result<Self>
where
Self: Sized;
@ -40,7 +40,7 @@ pub trait DatabaseEngine: Send + Sync {
}
}
pub trait Tree: Send + Sync {
pub trait KeyValueTree: Send + Sync {
fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>>;
fn insert(&self, key: &[u8], value: &[u8]) -> Result<()>;

740
src/database/key_value.rs
View File

@ -1,10 +1,7 @@
pub trait Data {
fn get_room_shortstatehash(room_id: &RoomId);
}
use crate::service;
/// Returns the last state hash key added to the db for the given room.
#[tracing::instrument(skip(self))]
pub fn current_shortstatehash(&self, room_id: &RoomId) -> Result<Option<u64>> {
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| {
@ -14,77 +11,21 @@ pub trait Data {
})
}
pub struct Service<D: Data> {
db: D,
}
impl Service {
/// Set the room to the given statehash and update caches.
#[tracing::instrument(skip(self, new_state_ids_compressed, db))]
pub fn force_state(
&self,
room_id: &RoomId,
shortstatehash: u64,
statediffnew :HashSet<CompressedStateEvent>,
statediffremoved :HashSet<CompressedStateEvent>,
db: &Database,
) -> Result<()> {
for event_id in statediffnew.into_iter().filter_map(|new| {
self.parse_compressed_state_event(new)
.ok()
.map(|(_, id)| id)
}) {
let pdu = match self.get_pdu_json(&event_id)? {
Some(pdu) => pdu,
None => continue,
};
if pdu.get("type").and_then(|val| val.as_str()) != Some("m.room.member") {
continue;
}
let pdu: PduEvent = match serde_json::from_str(
&serde_json::to_string(&pdu).expect("CanonicalJsonObj can be serialized to JSON"),
) {
Ok(pdu) => pdu,
Err(_) => continue,
};
#[derive(Deserialize)]
struct ExtractMembership {
membership: MembershipState,
}
let membership = match serde_json::from_str::<ExtractMembership>(pdu.content.get()) {
Ok(e) => e.membership,
Err(_) => continue,
};
let state_key = match pdu.state_key {
Some(k) => k,
None => continue,
};
let user_id = match UserId::parse(state_key) {
Ok(id) => id,
Err(_) => continue,
};
self.update_membership(room_id, &user_id, membership, &pdu.sender, None, db, false)?;
}
self.update_joined_count(room_id, db)?;
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(())
}
/// Returns the leaf pdus of a room.
#[tracing::instrument(skip(self))]
pub fn get_pdu_leaves(&self, room_id: &RoomId) -> Result<HashSet<Arc<EventId>>> {
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);
@ -99,15 +40,11 @@ impl Service {
.collect()
}
/// Replace the leaves of a room.
///
/// The provided `event_ids` become the new leaves, this allows a room to have multiple
/// `prev_events`.
#[tracing::instrument(skip(self))]
pub fn replace_pdu_leaves<'a>(
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);
@ -125,230 +62,48 @@ impl Service {
Ok(())
}
/// Generates a new StateHash and associates it with the incoming event.
///
/// This adds all current state events (not including the incoming event)
/// to `stateid_pduid` and adds the incoming event to `eventid_statehash`.
#[tracing::instrument(skip(self, state_ids_compressed, globals))]
pub fn set_event_state(
&self,
event_id: &EventId,
room_id: &RoomId,
state_ids_compressed: HashSet<CompressedStateEvent>,
globals: &super::globals::Globals,
) -> Result<()> {
let shorteventid = self.get_or_create_shorteventid(event_id, globals)?;
let previous_shortstatehash = self.current_shortstatehash(room_id)?;
let state_hash = self.calculate_hash(
&state_ids_compressed
.iter()
.map(|s| &s[..])
.collect::<Vec<_>>(),
);
let (shortstatehash, already_existed) =
self.get_or_create_shortstatehash(&state_hash, globals)?;
if !already_existed {
let states_parents = previous_shortstatehash
.map_or_else(|| Ok(Vec::new()), |p| self.load_shortstatehash_info(p))?;
let (statediffnew, statediffremoved) =
if let Some(parent_stateinfo) = states_parents.last() {
let statediffnew: HashSet<_> = state_ids_compressed
.difference(&parent_stateinfo.1)
.copied()
.collect();
let statediffremoved: HashSet<_> = parent_stateinfo
.1
.difference(&state_ids_compressed)
.copied()
.collect();
(statediffnew, statediffremoved)
} else {
(state_ids_compressed, HashSet::new())
};
self.save_state_from_diff(
shortstatehash,
statediffnew,
statediffremoved,
1_000_000, // high number because no state will be based on this one
states_parents,
)?;
}
self.shorteventid_shortstatehash
.insert(&shorteventid.to_be_bytes(), &shortstatehash.to_be_bytes())?;
Ok(())
}
/// Generates a new StateHash and associates it with the incoming event.
///
/// This adds all current state events (not including the incoming event)
/// to `stateid_pduid` and adds the incoming event to `eventid_statehash`.
#[tracing::instrument(skip(self, new_pdu, globals))]
pub fn append_to_state(
&self,
new_pdu: &PduEvent,
globals: &super::globals::Globals,
) -> Result<u64> {
let shorteventid = self.get_or_create_shorteventid(&new_pdu.event_id, globals)?;
let previous_shortstatehash = self.current_shortstatehash(&new_pdu.room_id)?;
if let Some(p) = previous_shortstatehash {
self.shorteventid_shortstatehash
.insert(&shorteventid.to_be_bytes(), &p.to_be_bytes())?;
}
if let Some(state_key) = &new_pdu.state_key {
let states_parents = previous_shortstatehash
.map_or_else(|| Ok(Vec::new()), |p| self.load_shortstatehash_info(p))?;
let shortstatekey = self.get_or_create_shortstatekey(
&new_pdu.kind.to_string().into(),
state_key,
globals,
)?;
let new = self.compress_state_event(shortstatekey, &new_pdu.event_id, globals)?;
let replaces = states_parents
.last()
.map(|info| {
info.1
.iter()
.find(|bytes| bytes.starts_with(&shortstatekey.to_be_bytes()))
})
.unwrap_or_default();
if Some(&new) == replaces {
return Ok(previous_shortstatehash.expect("must exist"));
}
// TODO: statehash with deterministic inputs
let shortstatehash = globals.next_count()?;
let mut statediffnew = HashSet::new();
statediffnew.insert(new);
let mut statediffremoved = HashSet::new();
if let Some(replaces) = replaces {
statediffremoved.insert(*replaces);
}
self.save_state_from_diff(
shortstatehash,
statediffnew,
statediffremoved,
2,
states_parents,
)?;
Ok(shortstatehash)
} else {
Ok(previous_shortstatehash.expect("first event in room must be a state event"))
}
}
#[tracing::instrument(skip(self, invite_event))]
pub fn calculate_invite_state(
&self,
invite_event: &PduEvent,
) -> Result<Vec<Raw<AnyStrippedStateEvent>>> {
let mut state = Vec::new();
// Add recommended events
if let Some(e) =
self.room_state_get(&invite_event.room_id, &StateEventType::RoomCreate, "")?
{
state.push(e.to_stripped_state_event());
}
if let Some(e) =
self.room_state_get(&invite_event.room_id, &StateEventType::RoomJoinRules, "")?
{
state.push(e.to_stripped_state_event());
}
if let Some(e) = self.room_state_get(
&invite_event.room_id,
&StateEventType::RoomCanonicalAlias,
"",
)? {
state.push(e.to_stripped_state_event());
}
if let Some(e) =
self.room_state_get(&invite_event.room_id, &StateEventType::RoomAvatar, "")?
{
state.push(e.to_stripped_state_event());
}
if let Some(e) =
self.room_state_get(&invite_event.room_id, &StateEventType::RoomName, "")?
{
state.push(e.to_stripped_state_event());
}
if let Some(e) = self.room_state_get(
&invite_event.room_id,
&StateEventType::RoomMember,
invite_event.sender.as_str(),
)? {
state.push(e.to_stripped_state_event());
}
state.push(invite_event.to_stripped_state_event());
Ok(state)
}
#[tracing::instrument(skip(self))]
pub fn set_room_state(&self, room_id: &RoomId, shortstatehash: u64) -> Result<()> {
self.roomid_shortstatehash
.insert(room_id.as_bytes(), &shortstatehash.to_be_bytes())?;
Ok(())
}
}
#[tracing::instrument(skip(self, globals))]
pub fn set_alias(
impl service::room::alias::Data for KeyValueDatabase {
fn set_alias(
&self,
alias: &RoomAliasId,
room_id: Option<&RoomId>,
globals: &super::globals::Globals,
room_id: Option<&RoomId>
) -> Result<()> {
if let Some(room_id) = room_id {
// New alias
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())?;
} else {
// room_id=None means remove alias
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.",
));
}
}
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(())
}
#[tracing::instrument(skip(self))]
pub fn id_from_alias(&self, alias: &RoomAliasId) -> Result<Option<Box<RoomId>>> {
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| {
@ -360,11 +115,10 @@ impl Service {
.transpose()
}
#[tracing::instrument(skip(self))]
pub fn room_aliases<'a>(
&'a self,
fn local_aliases_for_room(
&self,
room_id: &RoomId,
) -> impl Iterator<Item = Result<Box<RoomAliasId>>> + 'a {
) -> Result<()> {
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
@ -375,26 +129,22 @@ impl Service {
.map_err(|_| Error::bad_database("Invalid alias in aliasid_alias."))
})
}
}
#[tracing::instrument(skip(self))]
pub fn set_public(&self, room_id: &RoomId, public: bool) -> Result<()> {
if public {
self.publicroomids.insert(room_id.as_bytes(), &[])?;
} else {
self.publicroomids.remove(room_id.as_bytes())?;
}
Ok(())
impl service::room::directory::Data for KeyValueDatabase {
fn set_public(&self, room_id: &RoomId) -> Result<()> {
self.publicroomids.insert(room_id.as_bytes(), &[])?;
}
#[tracing::instrument(skip(self))]
pub fn is_public_room(&self, room_id: &RoomId) -> Result<bool> {
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())
}
#[tracing::instrument(skip(self))]
pub fn public_rooms(&self) -> impl Iterator<Item = Result<Box<RoomId>>> + '_ {
fn public_rooms(&self) -> impl Iterator<Item = Result<Box<RoomId>>> + '_ {
self.publicroomids.iter().map(|(bytes, _)| {
RoomId::parse(
utils::string_from_bytes(&bytes).map_err(|_| {
@ -404,43 +154,14 @@ impl Service {
.map_err(|_| Error::bad_database("Room ID in publicroomids is invalid."))
})
}
use crate::{database::abstraction::Tree, utils, Error, Result};
use ruma::{
events::{
presence::{PresenceEvent, PresenceEventContent},
receipt::ReceiptEvent,
SyncEphemeralRoomEvent,
},
presence::PresenceState,
serde::Raw,
signatures::CanonicalJsonObject,
RoomId, UInt, UserId,
};
use std::{
collections::{HashMap, HashSet},
mem,
sync::Arc,
};
pub struct RoomEdus {
pub(in super::super) readreceiptid_readreceipt: Arc<dyn Tree>, // ReadReceiptId = RoomId + Count + UserId
pub(in super::super) roomuserid_privateread: Arc<dyn Tree>, // RoomUserId = Room + User, PrivateRead = Count
pub(in super::super) roomuserid_lastprivatereadupdate: Arc<dyn Tree>, // LastPrivateReadUpdate = Count
pub(in super::super) typingid_userid: Arc<dyn Tree>, // TypingId = RoomId + TimeoutTime + Count
pub(in super::super) roomid_lasttypingupdate: Arc<dyn Tree>, // LastRoomTypingUpdate = Count
pub(in super::super) presenceid_presence: Arc<dyn Tree>, // PresenceId = RoomId + Count + UserId
pub(in super::super) userid_lastpresenceupdate: Arc<dyn Tree>, // LastPresenceUpdate = Count
}
impl RoomEdus {
/// Adds an event which will be saved until a new event replaces it (e.g. read receipt).
pub fn readreceipt_update(
impl service::room::edus::Data for KeyValueDatabase {
fn readreceipt_update(
&self,
user_id: &UserId,
room_id: &RoomId,
event: ReceiptEvent,
globals: &super::super::globals::Globals,
) -> Result<()> {
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
@ -477,8 +198,6 @@ impl RoomEdus {
Ok(())
}
/// Returns an iterator over the most recent read_receipts in a room that happened after the event with id `since`.
#[tracing::instrument(skip(self))]
pub fn readreceipts_since<'a>(
&'a self,
room_id: &RoomId,
@ -527,14 +246,11 @@ impl RoomEdus {
})
}
/// Sets a private read marker at `count`.
#[tracing::instrument(skip(self, globals))]
pub fn private_read_set(
fn private_read_set(
&self,
room_id: &RoomId,
user_id: &UserId,
count: u64,
globals: &super::super::globals::Globals,
) -> Result<()> {
let mut key = room_id.as_bytes().to_vec();
key.push(0xff);
@ -545,13 +261,9 @@ impl RoomEdus {
self.roomuserid_lastprivatereadupdate
.insert(&key, &globals.next_count()?.to_be_bytes())?;
Ok(())
}
/// Returns the private read marker.
#[tracing::instrument(skip(self))]
pub fn private_read_get(&self, room_id: &RoomId, user_id: &UserId) -> Result<Option<u64>> {
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());
@ -565,8 +277,7 @@ impl RoomEdus {
})
}
/// Returns the count of the last typing update in this room.
pub fn last_privateread_update(&self, user_id: &UserId, room_id: &RoomId) -> Result<u64> {
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());
@ -583,9 +294,7 @@ impl RoomEdus {
.unwrap_or(0))
}
/// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is
/// called.
pub fn typing_add(
fn typing_add(
&self,
user_id: &UserId,
room_id: &RoomId,
@ -611,12 +320,10 @@ impl RoomEdus {
Ok(())
}
/// Removes a user from typing before the timeout is reached.
pub fn typing_remove(
fn typing_remove(
&self,
user_id: &UserId,
room_id: &RoomId,
globals: &super::super::globals::Globals,
) -> Result<()> {
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
@ -643,59 +350,10 @@ impl RoomEdus {
Ok(())
}
/// Makes sure that typing events with old timestamps get removed.
fn typings_maintain(
fn last_typing_update(
&self,
room_id: &RoomId,
globals: &super::super::globals::Globals,
) -> Result<()> {
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
let current_timestamp = utils::millis_since_unix_epoch();
let mut found_outdated = false;
// Find all outdated edus before inserting a new one
for outdated_edu in self
.typingid_userid
.scan_prefix(prefix)
.map(|(key, _)| {
Ok::<_, Error>((
key.clone(),
utils::u64_from_bytes(
&key.splitn(2, |&b| b == 0xff).nth(1).ok_or_else(|| {
Error::bad_database("RoomTyping has invalid timestamp or delimiters.")
})?[0..mem::size_of::<u64>()],
)
.map_err(|_| Error::bad_database("RoomTyping has invalid timestamp bytes."))?,
))
})
.filter_map(|r| r.ok())
.take_while(|&(_, timestamp)| timestamp < current_timestamp)
{
// This is an outdated edu (time > timestamp)
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(())
}
/// Returns the count of the last typing update in this room.
#[tracing::instrument(skip(self, globals))]
pub fn last_typing_update(
&self,
room_id: &RoomId,
globals: &super::super::globals::Globals,
) -> Result<u64> {
self.typings_maintain(room_id, globals)?;
Ok(self
.roomid_lasttypingupdate
.get(room_id.as_bytes())?
@ -708,10 +366,10 @@ impl RoomEdus {
.unwrap_or(0))
}
pub fn typings_all(
fn typings_all(
&self,
room_id: &RoomId,
) -> Result<SyncEphemeralRoomEvent<ruma::events::typing::TypingEventContent>> {
) -> Result<HashSet<UserId>> {
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
@ -726,23 +384,14 @@ impl RoomEdus {
user_ids.insert(user_id);
}
Ok(SyncEphemeralRoomEvent {
content: ruma::events::typing::TypingEventContent {
user_ids: user_ids.into_iter().collect(),
},
})
Ok(user_ids)
}
/// Adds a presence event which will be saved until a new event replaces it.
///
/// Note: This method takes a RoomId because presence updates are always bound to rooms to
/// make sure users outside these rooms can't see them.
pub fn update_presence(
fn update_presence(
&self,
user_id: &UserId,
room_id: &RoomId,
presence: PresenceEvent,
globals: &super::super::globals::Globals,
) -> Result<()> {
// TODO: Remove old entry? Or maybe just wipe completely from time to time?
@ -767,8 +416,6 @@ impl RoomEdus {
Ok(())
}
/// Resets the presence timeout, so the user will stay in their current presence state.
#[tracing::instrument(skip(self))]
pub fn ping_presence(&self, user_id: &UserId) -> Result<()> {
self.userid_lastpresenceupdate.insert(
user_id.as_bytes(),
@ -778,8 +425,7 @@ impl RoomEdus {
Ok(())
}
/// Returns the timestamp of the last presence update of this user in millis since the unix epoch.
pub fn last_presence_update(&self, user_id: &UserId) -> Result<Option<u64>> {
fn last_presence_update(&self, user_id: &UserId) -> Result<Option<u64>> {
self.userid_lastpresenceupdate
.get(user_id.as_bytes())?
.map(|bytes| {
@ -790,125 +436,29 @@ impl RoomEdus {
.transpose()
}
pub fn get_last_presence_event(
fn get_presence_event(
&self,
user_id: &UserId,
room_id: &RoomId,
count: u64,
) -> Result<Option<PresenceEvent>> {
let last_update = match self.last_presence_update(user_id)? {
Some(last) => last,
None => return Ok(None),
};
let mut presence_id = room_id.as_bytes().to_vec();
presence_id.push(0xff);
presence_id.extend_from_slice(&last_update.to_be_bytes());
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| {
let mut presence: PresenceEvent = serde_json::from_slice(&value)
.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);
}
Ok(presence)
})
.map(|value| parse_presence_event(&value))
.transpose()
}
/// Sets all users to offline who have been quiet for too long.
fn _presence_maintain(
&self,
rooms: &super::Rooms,
globals: &super::super::globals::Globals,
) -> Result<()> {
let current_timestamp = utils::millis_since_unix_epoch();
for (user_id_bytes, last_timestamp) in self
.userid_lastpresenceupdate
.iter()
.filter_map(|(k, bytes)| {
Some((
k,
utils::u64_from_bytes(&bytes)
.map_err(|_| {
Error::bad_database("Invalid timestamp in userid_lastpresenceupdate.")
})
.ok()?,
))
})
.take_while(|(_, timestamp)| current_timestamp.saturating_sub(*timestamp) > 5 * 60_000)
// 5 Minutes
{
// Send new presence events to set the user offline
let count = globals.next_count()?.to_be_bytes();
let user_id: Box<_> = utils::string_from_bytes(&user_id_bytes)
.map_err(|_| {
Error::bad_database("Invalid UserId bytes in userid_lastpresenceupdate.")
})?
.try_into()
.map_err(|_| Error::bad_database("Invalid UserId in userid_lastpresenceupdate."))?;
for room_id in rooms.rooms_joined(&user_id).filter_map(|r| r.ok()) {
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(&user_id_bytes);
self.presenceid_presence.insert(
&presence_id,
&serde_json::to_vec(&PresenceEvent {
content: PresenceEventContent {
avatar_url: None,
currently_active: None,
displayname: None,
last_active_ago: Some(
last_timestamp.try_into().expect("time is valid"),
),
presence: PresenceState::Offline,
status_msg: None,
},
sender: user_id.to_owned(),
})
.expect("PresenceEvent can be serialized"),
)?;
}
self.userid_lastpresenceupdate.insert(
user_id.as_bytes(),
&utils::millis_since_unix_epoch().to_be_bytes(),
)?;
}
Ok(())
}
/// Returns an iterator over the most recent presence updates that happened after the event with id `since`.
#[tracing::instrument(skip(self, since, _rooms, _globals))]
pub fn presence_since(
fn presence_since(
&self,
room_id: &RoomId,
since: u64,
_rooms: &super::Rooms,
_globals: &super::super::globals::Globals,
) -> Result<HashMap<Box<UserId>, PresenceEvent>> {
//self.presence_maintain(rooms, globals)?;
let mut prefix = room_id.as_bytes().to_vec();
prefix.push(0xff);
@ -931,23 +481,7 @@ impl RoomEdus {
)
.map_err(|_| Error::bad_database("Invalid UserId in presenceid_presence."))?;
let mut presence: PresenceEvent = serde_json::from_slice(&value)
.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);
}
let presence = parse_presence_event(&value)?;
hashmap.insert(user_id, presence);
}
@ -956,8 +490,28 @@ impl RoomEdus {
}
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_was_sent_before(
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);
}
}
impl service::room::lazy_load::Data for KeyValueDatabase {
fn lazy_load_was_sent_before(
&self,
user_id: &UserId,
device_id: &DeviceId,
@ -974,28 +528,7 @@ impl RoomEdus {
Ok(self.lazyloadedids.get(&key)?.is_some())
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_mark_sent(
&self,
user_id: &UserId,
device_id: &DeviceId,
room_id: &RoomId,
lazy_load: HashSet<Box<UserId>>,
count: u64,
) {
self.lazy_load_waiting.lock().unwrap().insert(
(
user_id.to_owned(),
device_id.to_owned(),
room_id.to_owned(),
count,
),
lazy_load,
);
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_confirm_delivery(
fn lazy_load_confirm_delivery(
&self,
user_id: &UserId,
device_id: &DeviceId,
@ -1025,8 +558,7 @@ impl RoomEdus {
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn lazy_load_reset(
fn lazy_load_reset(
&self,
user_id: &UserId,
device_id: &DeviceId,
@ -1045,10 +577,10 @@ impl RoomEdus {
Ok(())
}
}
/// Checks if a room exists.
#[tracing::instrument(skip(self))]
pub fn exists(&self, room_id: &RoomId) -> Result<bool> {
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),
@ -1062,36 +594,10 @@ impl RoomEdus {
.filter(|(k, _)| k.starts_with(&prefix))
.is_some())
}
}
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
}
})
}
/// Returns the pdu from the outlier tree.
pub fn get_outlier_pdu_json(&self, event_id: &EventId) -> Result<Option<CanonicalJsonObject>> {
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| {
@ -1099,8 +605,7 @@ impl RoomEdus {
})
}
/// Returns the pdu from the outlier tree.
pub fn get_pdu_outlier(&self, event_id: &EventId) -> Result<Option<PduEvent>> {
fn get_outlier_pdu(&self, event_id: &EventId) -> Result<Option<PduEvent>> {
self.eventid_outlierpdu
.get(event_id.as_bytes())?
.map_or(Ok(None), |pdu| {
@ -1108,18 +613,16 @@ impl RoomEdus {
})
}
/// Append the PDU as an outlier.
#[tracing::instrument(skip(self, pdu))]
pub fn add_pdu_outlier(&self, event_id: &EventId, pdu: &CanonicalJsonObject) -> Result<()> {
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"),
)
}
}
#[tracing::instrument(skip(self, room_id, event_ids))]
pub fn mark_as_referenced(&self, room_id: &RoomId, event_ids: &[Arc<EventId>]) -> Result<()> {
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());
@ -1129,22 +632,19 @@ impl RoomEdus {
Ok(())
}
#[tracing::instrument(skip(self))]
pub fn is_event_referenced(&self, room_id: &RoomId, event_id: &EventId) -> Result<bool> {
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())
}
#[tracing::instrument(skip(self))]
pub fn mark_event_soft_failed(&self, event_id: &EventId) -> Result<()> {
fn mark_event_soft_failed(&self, event_id: &EventId) -> Result<()> {
self.softfailedeventids.insert(event_id.as_bytes(), &[])
}
#[tracing::instrument(skip(self))]
pub fn is_event_soft_failed(&self, event_id: &EventId) -> Result<bool> {
fn is_event_soft_failed(&self, event_id: &EventId) -> Result<bool> {
self.softfailedeventids
.get(event_id.as_bytes())
.map(|o| o.is_some())
}
}

48
src/database/mod.rs
View File

@ -15,7 +15,7 @@ pub mod users;
use self::admin::create_admin_room;
use crate::{utils, Config, Error, Result};
use abstraction::DatabaseEngine;
use abstraction::KeyValueDatabaseEngine;
use directories::ProjectDirs;
use futures_util::{stream::FuturesUnordered, StreamExt};
use lru_cache::LruCache;
@ -39,8 +39,8 @@ use std::{
use tokio::sync::{mpsc, OwnedRwLockReadGuard, RwLock as TokioRwLock, Semaphore};
use tracing::{debug, error, info, warn};
pub struct Database {
_db: Arc<dyn DatabaseEngine>,
pub struct KeyValueDatabase {
_db: Arc<dyn KeyValueDatabaseEngine>,
pub globals: globals::Globals,
pub users: users::Users,
pub uiaa: uiaa::Uiaa,
@ -55,7 +55,7 @@ pub struct Database {
pub pusher: pusher::PushData,
}
impl Database {
impl KeyValueDatabase {
/// Tries to remove the old database but ignores all errors.
pub fn try_remove(server_name: &str) -> Result<()> {
let mut path = ProjectDirs::from("xyz", "koesters", "conduit")
@ -124,7 +124,7 @@ impl Database {
.map_err(|_| Error::BadConfig("Database folder doesn't exists and couldn't be created (e.g. due to missing permissions). Please create the database folder yourself."))?;
}
let builder: Arc<dyn DatabaseEngine> = match &*config.database_backend {
let builder: Arc<dyn KeyValueDatabaseEngine> = match &*config.database_backend {
"sqlite" => {
#[cfg(not(feature = "sqlite"))]
return Err(Error::BadConfig("Database backend not found."));
@ -955,7 +955,7 @@ impl Database {
}
/// Sets the emergency password and push rules for the @conduit account in case emergency password is set
fn set_emergency_access(db: &Database) -> Result<bool> {
fn set_emergency_access(db: &KeyValueDatabase) -> Result<bool> {
let conduit_user = UserId::parse_with_server_name("conduit", db.globals.server_name())
.expect("@conduit:server_name is a valid UserId");
@ -979,39 +979,3 @@ fn set_emergency_access(db: &Database) -> Result<bool> {
res
}
pub struct DatabaseGuard(OwnedRwLockReadGuard<Database>);
impl Deref for DatabaseGuard {
type Target = OwnedRwLockReadGuard<Database>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
#[cfg(feature = "conduit_bin")]
#[axum::async_trait]
impl<B> axum::extract::FromRequest<B> for DatabaseGuard
where
B: Send,
{
type Rejection = axum::extract::rejection::ExtensionRejection;
async fn from_request(
req: &mut axum::extract::RequestParts<B>,
) -> Result<Self, Self::Rejection> {
use axum::extract::Extension;
let Extension(db): Extension<Arc<TokioRwLock<Database>>> =
Extension::from_request(req).await?;
Ok(DatabaseGuard(db.read_owned().await))
}
}
impl From<OwnedRwLockReadGuard<Database>> for DatabaseGuard {
fn from(val: OwnedRwLockReadGuard<Database>) -> Self {
Self(val)
}
}

49
src/main.rs
View File

@ -46,27 +46,26 @@ use tikv_jemallocator::Jemalloc;
#[global_allocator]
static GLOBAL: Jemalloc = Jemalloc;
#[tokio::main]
async fn main() {
let raw_config =
Figment::new()
.merge(
Toml::file(Env::var("CONDUIT_CONFIG").expect(
"The CONDUIT_CONFIG env var needs to be set. Example: /etc/conduit.toml",
))
.nested(),
)
.merge(Env::prefixed("CONDUIT_").global());
lazy_static! {
static ref DB: Database = {
let raw_config =
Figment::new()
.merge(
Toml::file(Env::var("CONDUIT_CONFIG").expect(
"The CONDUIT_CONFIG env var needs to be set. Example: /etc/conduit.toml",
))
.nested(),
)
.merge(Env::prefixed("CONDUIT_").global());
let config = match raw_config.extract::<Config>() {
Ok(s) => s,
Err(e) => {
eprintln!("It looks like your config is invalid. The following error occured while parsing it: {}", e);
std::process::exit(1);
}
};
let config = match raw_config.extract::<Config>() {
Ok(s) => s,
Err(e) => {
eprintln!("It looks like your config is invalid. The following error occured while parsing it: {}", e);
std::process::exit(1);
}
};
let start = async {
config.warn_deprecated();
let db = match Database::load_or_create(&config).await {
@ -79,8 +78,15 @@ async fn main() {
std::process::exit(1);
}
};
};
}
run_server(&config, db).await.unwrap();
#[tokio::main]
async fn main() {
lazy_static::initialize(&DB);
let start = async {
run_server(&config).await.unwrap();
};
if config.allow_jaeger {
@ -120,7 +126,8 @@ async fn main() {
}
}
async fn run_server(config: &Config, db: Arc<RwLock<Database>>) -> io::Result<()> {
async fn run_server() -> io::Result<()> {
let config = DB.globals.config;
let addr = SocketAddr::from((config.address, config.port));
let x_requested_with = HeaderName::from_static("x-requested-with");

22
src/service/rooms/alias/data.rs Normal file
View File

@ -0,0 +1,22 @@
pub trait Data {
/// Creates or updates the alias to the given room id.
pub fn set_alias(
alias: &RoomAliasId,
room_id: &RoomId
) -> Result<()>;
/// Forgets about an alias. Returns an error if the alias did not exist.
pub fn remove_alias(
alias: &RoomAliasId,
) -> Result<()>;
/// Looks up the roomid for the given alias.
pub fn resolve_local_alias(
alias: &RoomAliasId,
) -> Result<()>;
/// Returns all local aliases that point to the given room
pub fn local_aliases_for_room(
alias: &RoomAliasId,
) -> Result<()>;
}

72
src/service/rooms/alias/mod.rs
View File

@ -1,66 +1,40 @@
mod data;
pub use data::Data;
use crate::service::*;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
#[tracing::instrument(skip(self, globals))]
pub fn set_alias(
&self,
alias: &RoomAliasId,
room_id: Option<&RoomId>,
globals: &super::globals::Globals,
room_id: &RoomId,
) -> Result<()> {
if let Some(room_id) = room_id {
// New alias
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())?;
} else {
// room_id=None means remove alias
if let Some(room_id) = self.alias_roomid.get(alias.alias().as_bytes())? {
let mut prefix = room_id.to_vec();
prefix.push(0xff);
self.db.set_alias(alias, room_id)
}
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(())
#[tracing::instrument(skip(self, globals))]
pub fn remove_alias(
&self,
alias: &RoomAliasId,
) -> Result<()> {
self.db.remove_alias(alias)
}
#[tracing::instrument(skip(self))]
pub fn id_from_alias(&self, alias: &RoomAliasId) -> Result<Option<Box<RoomId>>> {
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()
pub fn resolve_local_alias(&self, alias: &RoomAliasId) -> Result<Option<Box<RoomId>>> {
self.db.resolve_local_alias(alias: &RoomAliasId)
}
#[tracing::instrument(skip(self))]
pub fn room_aliases<'a>(
pub fn local_aliases_for_room<'a>(
&'a self,
room_id: &RoomId,
) -> impl Iterator<Item = Result<Box<RoomAliasId>>> + 'a {
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."))
})
self.db.local_aliases_for_room(room_id)
}
}

13
src/service/rooms/directory/data.rs Normal file
View File

@ -0,0 +1,13 @@
pub trait Data {
/// Adds the room to the public room directory
fn set_public(room_id: &RoomId) -> Result<()>;
/// Removes the room from the public room directory.
fn set_not_public(room_id: &RoomId) -> Result<()>;
/// Returns true if the room is in the public room directory.
fn is_public_room(room_id: &RoomId) -> Result<bool>;
/// Returns the unsorted public room directory
fn public_rooms() -> impl Iterator<Item = Result<Box<RoomId>>> + '_;
}

37
src/service/rooms/directory/mod.rs
View File

@ -1,29 +1,30 @@
mod data;
pub use data::Data;
use crate::service::*;
pub struct Service<D: Data> {
db: D,
}
impl Service<_> {
#[tracing::instrument(skip(self))]
pub fn set_public(&self, room_id: &RoomId) -> Result<()> {
self.db.set_public(&self, room_id)
}
#[tracing::instrument(skip(self))]
pub fn set_public(&self, room_id: &RoomId, public: bool) -> Result<()> {
if public {
self.publicroomids.insert(room_id.as_bytes(), &[])?;
} else {
self.publicroomids.remove(room_id.as_bytes())?;
}
Ok(())
pub fn set_not_public(&self, room_id: &RoomId) -> Result<()> {
self.db.set_not_public(&self, room_id)
}
#[tracing::instrument(skip(self))]
pub fn is_public_room(&self, room_id: &RoomId) -> Result<bool> {
Ok(self.publicroomids.get(room_id.as_bytes())?.is_some())
self.db.is_public_room(&self, room_id)
}
#[tracing::instrument(skip(self))]
pub 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."))
})
self.db.public_rooms(&self, room_id)
}
}

91
src/service/rooms/edus/data.rs Normal file
View File

@ -0,0 +1,91 @@
pub trait Data {
/// Replaces the previous read receipt.
fn readreceipt_update(
&self,
user_id: &UserId,
room_id: &RoomId,
event: ReceiptEvent,
) -> Result<()>;
/// Returns an iterator over the most recent read_receipts in a room that happened after the event with id `since`.
fn readreceipts_since(
&self,
room_id: &RoomId,
since: u64,
) -> impl Iterator<
Item = Result<(
Box<UserId>,
u64,
Raw<ruma::events::AnySyncEphemeralRoomEvent>,
)>,
>;
/// Sets a private read marker at `count`.
fn private_read_set(
&self,
room_id: &RoomId,
user_id: &UserId,
count: u64,
) -> Result<()>;
/// Returns the private read marker.
fn private_read_get(&self, room_id: &RoomId, user_id: &UserId) -> Result<Option<u64>>;
/// Returns the count of the last typing update in this room.
fn last_privateread_update(&self, user_id: &UserId, room_id: &RoomId) -> Result<u64>;
/// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is
/// called.
fn typing_add(
&self,
user_id: &UserId,
room_id: &RoomId,
timeout: u64,
) -> Result<()>;
/// Removes a user from typing before the timeout is reached.
fn typing_remove(
&self,
user_id: &UserId,
room_id: &RoomId,
) -> Result<()>;
/// Returns the count of the last typing update in this room.
fn last_typing_update(
&self,
room_id: &RoomId,
) -> Result<u64>;
/// Returns all user ids currently typing.
fn typings_all(
&self,
room_id: &RoomId,
) -> Result<HashSet<UserId>>;
/// Adds a presence event which will be saved until a new event replaces it.
///
/// Note: This method takes a RoomId because presence updates are always bound to rooms to
/// make sure users outside these rooms can't see them.
fn update_presence(
&self,
user_id: &UserId,
room_id: &RoomId,
presence: PresenceEvent,
) -> Result<()>;
/// Resets the presence timeout, so the user will stay in their current presence state.
fn ping_presence(&self, user_id: &UserId) -> Result<()>;
/// Returns the timestamp of the last presence update of this user in millis since the unix epoch.
fn last_presence_update(&self, user_id: &UserId) -> Result<Option<u64>>;
/// Returns the presence event with correct last_active_ago.
fn get_presence_event(&self, room_id: &RoomId, user_id: &UserId, count: u64) -> Result<Option<PresenceEvent>>;
/// Returns the most recent presence updates that happened after the event with id `since`.
fn presence_since(
&self,
room_id: &RoomId,
since: u64,
) -> Result<HashMap<Box<UserId>, PresenceEvent>>;
}

350
src/service/rooms/edus/mod.rs
View File

@ -1,73 +1,21 @@
use crate::{database::abstraction::Tree, utils, Error, Result};
use ruma::{
events::{
presence::{PresenceEvent, PresenceEventContent},
receipt::ReceiptEvent,
SyncEphemeralRoomEvent,
},
presence::PresenceState,
serde::Raw,
signatures::CanonicalJsonObject,
RoomId, UInt, UserId,
};
use std::{
collections::{HashMap, HashSet},
mem,
sync::Arc,
};
mod data;
pub use data::Data;
pub struct RoomEdus {
pub(in super::super) readreceiptid_readreceipt: Arc<dyn Tree>, // ReadReceiptId = RoomId + Count + UserId
pub(in super::super) roomuserid_privateread: Arc<dyn Tree>, // RoomUserId = Room + User, PrivateRead = Count
pub(in super::super) roomuserid_lastprivatereadupdate: Arc<dyn Tree>, // LastPrivateReadUpdate = Count
pub(in super::super) typingid_userid: Arc<dyn Tree>, // TypingId = RoomId + TimeoutTime + Count
pub(in super::super) roomid_lasttypingupdate: Arc<dyn Tree>, // LastRoomTypingUpdate = Count
pub(in super::super) presenceid_presence: Arc<dyn Tree>, // PresenceId = RoomId + Count + UserId
pub(in super::super) userid_lastpresenceupdate: Arc<dyn Tree>, // LastPresenceUpdate = Count
use crate::service::*;
pub struct Service<D: Data> {
db: D,
}
impl RoomEdus {
/// Adds an event which will be saved until a new event replaces it (e.g. read receipt).
impl Service<_> {
/// Replaces the previous read receipt.
pub fn readreceipt_update(
&self,
user_id: &UserId,
room_id: &RoomId,
event: ReceiptEvent,
globals: &super::super::globals::Globals,
) -> 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(())
self.db.readreceipt_update(user_id, room_id, event);
}
/// Returns an iterator over the most recent read_receipts in a room that happened after the event with id `since`.
@ -83,41 +31,7 @@ impl RoomEdus {
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"),
),
))
})
self.db.readreceipts_since(room_id, since)
}
/// Sets a private read marker at `count`.
@ -127,53 +41,19 @@ impl RoomEdus {
room_id: &RoomId,
user_id: &UserId,
count: u64,
globals: &super::super::globals::Globals,
) -> 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())?;
Ok(())
self.db.private_read_set(room_id, user_id, count)
}
/// Returns the private read marker.
#[tracing::instrument(skip(self))]
pub 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")
})?))
})
self.db.private_read_get(room_id, user_id)
}
/// Returns the count of the last typing update in this room.
pub 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))
self.db.last_privateread_update(user_id, room_id)
}
/// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is
@ -183,25 +63,8 @@ impl RoomEdus {
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(())
self.db.typing_add(user_id, room_id, timeout)
}
/// Removes a user from typing before the timeout is reached.
@ -209,33 +72,11 @@ impl RoomEdus {
&self,
user_id: &UserId,
room_id: &RoomId,
globals: &super::super::globals::Globals,
) -> 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(())
self.db.typing_remove(user_id, room_id)
}
/* TODO: Do this in background thread?
/// Makes sure that typing events with old timestamps get removed.
fn typings_maintain(
&self,
@ -279,45 +120,23 @@ impl RoomEdus {
Ok(())
}
*/
/// Returns the count of the last typing update in this room.
#[tracing::instrument(skip(self, globals))]
pub fn last_typing_update(
&self,
room_id: &RoomId,
globals: &super::super::globals::Globals,
) -> Result<u64> {
self.typings_maintain(room_id, globals)?;
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))
self.db.last_typing_update(room_id)
}
/// Returns a new typing EDU.
pub fn typings_all(
&self,
room_id: &RoomId,
) -> Result<SyncEphemeralRoomEvent<ruma::events::typing::TypingEventContent>> {
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);
}
let user_ids = self.db.typings_all(room_id)?;
Ok(SyncEphemeralRoomEvent {
content: ruma::events::typing::TypingEventContent {
@ -335,52 +154,13 @@ impl RoomEdus {
user_id: &UserId,
room_id: &RoomId,
presence: PresenceEvent,
globals: &super::super::globals::Globals,
) -> 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(())
self.db.update_presence(user_id, room_id, presence)
}
/// Resets the presence timeout, so the user will stay in their current presence state.
#[tracing::instrument(skip(self))]
pub 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(())
}
/// Returns the timestamp of the last presence update of this user in millis since the unix epoch.
pub 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()
self.db.ping_presence(user_id)
}
pub fn get_last_presence_event(
@ -388,42 +168,15 @@ impl RoomEdus {
user_id: &UserId,
room_id: &RoomId,
) -> Result<Option<PresenceEvent>> {
let last_update = match self.last_presence_update(user_id)? {
let last_update = match self.db.last_presence_update(user_id)? {
Some(last) => last,
None => return Ok(None),
};
let mut presence_id = room_id.as_bytes().to_vec();
presence_id.push(0xff);
presence_id.extend_from_slice(&last_update.to_be_bytes());
presence_id.push(0xff);
presence_id.extend_from_slice(user_id.as_bytes());
self.presenceid_presence
.get(&presence_id)?
.map(|value| {
let mut presence: PresenceEvent = serde_json::from_slice(&value)
.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);
}
Ok(presence)
})
.transpose()
self.db.get_presence_event(room_id, user_id, last_update)
}
/* TODO
/// Sets all users to offline who have been quiet for too long.
fn _presence_maintain(
&self,
@ -489,62 +242,15 @@ impl RoomEdus {
}
Ok(())
}
}*/
/// Returns an iterator over the most recent presence updates that happened after the event with id `since`.
/// Returns the most recent presence updates that happened after the event with id `since`.
#[tracing::instrument(skip(self, since, _rooms, _globals))]
pub fn presence_since(
&self,
room_id: &RoomId,
since: u64,
_rooms: &super::Rooms,
_globals: &super::super::globals::Globals,
) -> Result<HashMap<Box<UserId>, PresenceEvent>> {
//self.presence_maintain(rooms, globals)?;
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 mut presence: PresenceEvent = serde_json::from_slice(&value)
.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);
}
hashmap.insert(user_id, presence);
}
Ok(hashmap)
self.db.presence_since(room_id, since)
}
}

2052
src/service/rooms/event_handler/mod.rs
View File

File diff suppressed because it is too large Load Diff

24
src/service/rooms/lazy_loading/data.rs Normal file
View File

@ -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<()>;
}

55
src/service/rooms/lazy_loading/mod.rs
View File

@ -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);
}
}

3
src/service/rooms/metadata/data.rs Normal file
View File

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

52
src/service/rooms/metadata/mod.rs
View File

@ -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
}
})
}
}

5
src/service/rooms/outlier/data.rs Normal file
View File

@ -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<()>;
}

29
src/service/rooms/outlier/mod.rs
View File

@ -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)
}
}

6
src/service/rooms/pdu_metadata/data.rs Normal file
View File

@ -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>;
}

29
src/service/rooms/pdu_metadata/mod.rs
View File

@ -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)
}
}

27
src/service/rooms/short/mod.rs
View File

@ -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
}
})
}

32
src/service/rooms/state/data.rs
View File

@ -1,16 +1,24 @@
pub trait Data {
/// Returns the last state hash key added to the db for the given room.
fn get_room_shortstatehash(room_id: &RoomId);
/// Update the current state of the room.
fn set_room_state(room_id: &RoomId, new_shortstatehash: u64
_mutex_lock: &MutexGuard<'_, StateLock>, // Take mutex guard to make sure users get the room state mutex
);
/// Associates a state with an event.
fn set_event_state(shorteventid: u64, shortstatehash: u64) -> Result<()> {
/// Returns all events we would send as the prev_events of the next event.
fn get_forward_extremities(room_id: &RoomId) -> Result<HashSet<Arc<EventId>>>;
/// Replace the forward extremities of the room.
fn set_forward_extremities(
room_id: &RoomId,
event_ids: impl IntoIterator<Item = &'_ EventId> + Debug,
_mutex_lock: &MutexGuard<'_, StateLock>, // Take mutex guard to make sure users get the room state mutex
) -> Result<()> {
}
/// Returns the last state hash key added to the db for the given room.
#[tracing::instrument(skip(self))]
pub fn current_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")
})?))
})
}
pub struct StateLock;

86
src/service/rooms/state/mod.rs
View File

@ -1,25 +1,30 @@
mod data;
pub use data::Data;
use crate::service::*;
pub struct Service<D: Data> {
db: D,
}
impl Service {
impl Service<_> {
/// Set the room to the given statehash and update caches.
#[tracing::instrument(skip(self, new_state_ids_compressed, db))]
pub fn force_state(
&self,
room_id: &RoomId,
shortstatehash: u64,
statediffnew :HashSet<CompressedStateEvent>,
statediffremoved :HashSet<CompressedStateEvent>,
statediffnew: HashSet<CompressedStateEvent>,
statediffremoved: HashSet<CompressedStateEvent>,
db: &Database,
) -> Result<()> {
for event_id in statediffnew.into_iter().filter_map(|new| {
self.parse_compressed_state_event(new)
state_compressor::parse_compressed_state_event(new)
.ok()
.map(|(_, id)| id)
}) {
let pdu = match self.get_pdu_json(&event_id)? {
let pdu = match timeline::get_pdu_json(&event_id)? {
Some(pdu) => pdu,
None => continue,
};
@ -55,56 +60,12 @@ impl Service {
Err(_) => continue,
};
self.update_membership(room_id, &user_id, membership, &pdu.sender, None, db, false)?;
room::state_cache::update_membership(room_id, &user_id, membership, &pdu.sender, None, db, false)?;
}
self.update_joined_count(room_id, db)?;
room::state_cache::update_joined_count(room_id, db)?;
self.roomid_shortstatehash
.insert(room_id.as_bytes(), &new_shortstatehash.to_be_bytes())?;
Ok(())
}
/// Returns the leaf pdus of a room.
#[tracing::instrument(skip(self))]
pub 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()
}
/// Replace the leaves of a room.
///
/// The provided `event_ids` become the new leaves, this allows a room to have multiple
/// `prev_events`.
#[tracing::instrument(skip(self))]
pub fn replace_pdu_leaves<'a>(
&self,
room_id: &RoomId,
event_ids: impl IntoIterator<Item = &'a EventId> + Debug,
) -> 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())?;
}
db.set_room_state(room_id, new_shortstatehash);
Ok(())
}
@ -121,11 +82,11 @@ impl Service {
state_ids_compressed: HashSet<CompressedStateEvent>,
globals: &super::globals::Globals,
) -> Result<()> {
let shorteventid = self.get_or_create_shorteventid(event_id, globals)?;
let shorteventid = short::get_or_create_shorteventid(event_id, globals)?;
let previous_shortstatehash = self.current_shortstatehash(room_id)?;
let previous_shortstatehash = db.get_room_shortstatehash(room_id)?;
let state_hash = self.calculate_hash(
let state_hash = super::calculate_hash(
&state_ids_compressed
.iter()
.map(|s| &s[..])
@ -133,11 +94,11 @@ impl Service {
);
let (shortstatehash, already_existed) =
self.get_or_create_shortstatehash(&state_hash, globals)?;
short::get_or_create_shortstatehash(&state_hash, globals)?;
if !already_existed {
let states_parents = previous_shortstatehash
.map_or_else(|| Ok(Vec::new()), |p| self.load_shortstatehash_info(p))?;
.map_or_else(|| Ok(Vec::new()), |p| room::state_compressor.load_shortstatehash_info(p))?;
let (statediffnew, statediffremoved) =
if let Some(parent_stateinfo) = states_parents.last() {
@ -156,7 +117,7 @@ impl Service {
} else {
(state_ids_compressed, HashSet::new())
};
self.save_state_from_diff(
state_compressor::save_state_from_diff(
shortstatehash,
statediffnew,
statediffremoved,
@ -165,8 +126,7 @@ impl Service {
)?;
}
self.shorteventid_shortstatehash
.insert(&shorteventid.to_be_bytes(), &shortstatehash.to_be_bytes())?;
db.set_event_state(&shorteventid.to_be_bytes(), &shortstatehash.to_be_bytes())?;
Ok(())
}
@ -183,7 +143,7 @@ impl Service {
) -> Result<u64> {
let shorteventid = self.get_or_create_shorteventid(&new_pdu.event_id, globals)?;
let previous_shortstatehash = self.current_shortstatehash(&new_pdu.room_id)?;
let previous_shortstatehash = self.get_room_shortstatehash(&new_pdu.room_id)?;
if let Some(p) = previous_shortstatehash {
self.shorteventid_shortstatehash
@ -293,4 +253,8 @@ impl Service {
Ok(())
}
pub fn db(&self) -> D {
&self.db
}
}