magmaus3/continuwuity
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge remote-tracking branch 'origin/master' into federation

Browse source
This commit is contained in:
Timo Kösters 2020-09-12 22:13:53 +02:00
commit e6b1f54857
No known key found for this signature in database
GPG key ID: 24DA7517711A2BA4
22 changed files with 925 additions and 169 deletions
Download patch file Download diff file Expand all files Collapse all files

154
src/database/key_backups.rs
View file

@ -37,6 +37,28 @@ impl KeyBackups {
Ok(version)
}
pub fn delete_backup(&self, user_id: &UserId, version: &str) -> Result<()> {
let mut key = user_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&version.as_bytes());
self.backupid_algorithm.remove(&key)?;
self.backupid_etag.remove(&key)?;
key.push(0xff);
for outdated_key in self
.backupkeyid_backup
.scan_prefix(&key)
.keys()
.filter_map(|r| r.ok())
{
self.backupkeyid_backup.remove(outdated_key)?;
}
Ok(())
}
pub fn update_backup(
&self,
user_id: &UserId,
@ -163,6 +185,7 @@ impl KeyBackups {
let mut prefix = user_id.to_string().as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(version.as_bytes());
prefix.push(0xff);
let mut rooms = BTreeMap::<RoomId, Sessions>::new();
@ -204,4 +227,135 @@ impl KeyBackups {
Ok(rooms)
}
pub fn get_room(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
) -> BTreeMap<String, KeyData> {
let mut prefix = user_id.to_string().as_bytes().to_vec();
prefix.push(0xff);
prefix.extend_from_slice(version.as_bytes());
prefix.push(0xff);
prefix.extend_from_slice(room_id.as_bytes());
prefix.push(0xff);
self.backupkeyid_backup
.scan_prefix(&prefix)
.map(|r| {
let (key, value) = r?;
let mut parts = key.rsplit(|&b| b == 0xff);
let session_id =
utils::string_from_bytes(&parts.next().ok_or_else(|| {
Error::bad_database("backupkeyid_backup key is invalid.")
})?)
.map_err(|_| {
Error::bad_database("backupkeyid_backup session_id is invalid.")
})?;
let key_data = serde_json::from_slice(&value).map_err(|_| {
Error::bad_database("KeyData in backupkeyid_backup is invalid.")
})?;
Ok::<_, Error>((session_id, key_data))
})
.filter_map(|r| r.ok())
.collect()
}
pub fn get_session(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
session_id: &str,
) -> Result<Option<KeyData>> {
let mut key = user_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(version.as_bytes());
key.push(0xff);
key.extend_from_slice(room_id.as_bytes());
key.push(0xff);
key.extend_from_slice(session_id.as_bytes());
self.backupkeyid_backup
.get(&key)?
.map(|value| {
serde_json::from_slice(&value)
.map_err(|_| Error::bad_database("KeyData in backupkeyid_backup is invalid."))
})
.transpose()
}
pub fn delete_all_keys(&self, user_id: &UserId, version: &str) -> Result<()> {
let mut key = user_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&version.as_bytes());
key.push(0xff);
for outdated_key in self
.backupkeyid_backup
.scan_prefix(&key)
.keys()
.filter_map(|r| r.ok())
{
self.backupkeyid_backup.remove(outdated_key)?;
}
Ok(())
}
pub fn delete_room_keys(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
) -> Result<()> {
let mut key = user_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&version.as_bytes());
key.push(0xff);
key.extend_from_slice(&room_id.as_bytes());
key.push(0xff);
for outdated_key in self
.backupkeyid_backup
.scan_prefix(&key)
.keys()
.filter_map(|r| r.ok())
{
self.backupkeyid_backup.remove(outdated_key)?;
}
Ok(())
}
pub fn delete_room_key(
&self,
user_id: &UserId,
version: &str,
room_id: &RoomId,
session_id: &str,
) -> Result<()> {
let mut key = user_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&version.as_bytes());
key.push(0xff);
key.extend_from_slice(&room_id.as_bytes());
key.push(0xff);
key.extend_from_slice(&session_id.as_bytes());
for outdated_key in self
.backupkeyid_backup
.scan_prefix(&key)
.keys()
.filter_map(|r| r.ok())
{
self.backupkeyid_backup.remove(outdated_key)?;
}
Ok(())
}
}

148
src/database/rooms.rs
View file

@ -12,7 +12,6 @@ use ruma::{
room::{
member,
power_levels::{self, PowerLevelsEventContent},
redaction,
},
EventType,
},
@ -47,6 +46,7 @@ pub struct Rooms {
pub(super) userroomid_joined: sled::Tree,
pub(super) roomuserid_joined: sled::Tree,
pub(super) roomuseroncejoinedids: sled::Tree,
pub(super) userroomid_invited: sled::Tree,
pub(super) roomuserid_invited: sled::Tree,
pub(super) userroomid_left: sled::Tree,
@ -477,30 +477,16 @@ impl Rooms {
self.append_to_state(&pdu_id, &pdu)?;
}
match &pdu.kind {
match pdu.kind {
EventType::RoomRedaction => {
if let Some(redact_id) = &pdu.redacts {
// TODO: Reason
let _reason = serde_json::from_value::<Raw<redaction::RedactionEventContent>>(
pdu.content,
)
.expect("Raw::from_value always works.")
.deserialize()
.map_err(|_| {
Error::BadRequest(
ErrorKind::InvalidParam,
"Invalid redaction event content.",
)
})?
.reason;
self.redact_pdu(&redact_id)?;
self.redact_pdu(&redact_id, &pdu)?;
}
}
EventType::RoomMember => {
if let Some(state_key) = pdu.state_key.as_ref() {
if let Some(state_key) = pdu.state_key {
// if the state_key fails
let target_user_id = UserId::try_from(state_key.as_str())
let target_user_id = UserId::try_from(state_key)
.expect("This state_key was previously validated");
// 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
@ -527,7 +513,7 @@ impl Rooms {
.split_terminator(|c: char| !c.is_alphanumeric())
.map(str::to_lowercase)
{
let mut key = pdu.room_id.as_bytes().to_vec();
let mut key = pdu.room_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(word.as_bytes());
key.push(0xff);
@ -538,7 +524,9 @@ impl Rooms {
}
_ => {}
}
self.edus.room_read_set(&pdu.room_id, &pdu.sender, index)?;
self.edus
.private_read_set(&pdu.room_id, &pdu.sender, index, &globals)?;
Ok(pdu.event_id)
}
@ -922,12 +910,12 @@ impl Rooms {
}
/// Replace a PDU with the redacted form.
pub fn redact_pdu(&self, event_id: &EventId) -> Result<()> {
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()?;
pdu.redact(&reason)?;
self.replace_pdu(&pdu_id, &pdu)?;
Ok(())
} else {
@ -959,6 +947,91 @@ impl Rooms {
match &membership {
member::MembershipState::Join => {
// Check if the user never joined this room
if !self.once_joined(&user_id, &room_id)? {
// Add the user ID to the join list then
self.roomuseroncejoinedids.insert(&userroom_id, &[])?;
// Check if the room has a predecessor
if let Some(predecessor) = serde_json::from_value::<
Raw<ruma::events::room::create::CreateEventContent>,
>(
self.room_state_get(&room_id, &EventType::RoomCreate, "")?
.ok_or_else(|| {
Error::bad_database("Found room without m.room.create event.")
})?
.content,
)
.expect("Raw::from_value always works")
.deserialize()
.map_err(|_| Error::bad_database("Invalid room event in database."))?
.predecessor
{
// Copy user settings from predecessor to the current room:
// - Push rules
//
// TODO: finish this once push rules are implemented.
//
// let mut push_rules_event_content = account_data
// .get::<ruma::events::push_rules::PushRulesEvent>(
// None,
// user_id,
// EventType::PushRules,
// )?;
//
// NOTE: find where `predecessor.room_id` match
// and update to `room_id`.
//
// account_data
// .update(
// None,
// user_id,
// EventType::PushRules,
// &push_rules_event_content,
// globals,
// )
// .ok();
// Copy old tags to new room
if let Some(tag_event) = account_data.get::<ruma::events::tag::TagEvent>(
Some(&predecessor.room_id),
user_id,
EventType::Tag,
)? {
account_data
.update(Some(room_id), user_id, EventType::Tag, &tag_event, globals)
.ok();
};
// Copy direct chat flag
if let Some(mut direct_event) = account_data
.get::<ruma::events::direct::DirectEvent>(
None,
user_id,
EventType::Direct,
)? {
let mut room_ids_updated = false;
for room_ids in direct_event.content.0.values_mut() {
if room_ids.iter().any(|r| r == &predecessor.room_id) {
room_ids.push(room_id.clone());
room_ids_updated = true;
}
}
if room_ids_updated {
account_data.update(
None,
user_id,
EventType::Direct,
&direct_event,
globals,
)?;
}
};
}
}
self.userroomid_joined.insert(&userroom_id, &[])?;
self.roomuserid_joined.insert(&roomuser_id, &[])?;
self.userroomid_invited.remove(&userroom_id)?;
@ -1218,6 +1291,27 @@ impl Rooms {
})
}
/// Returns an iterator over all User IDs who ever joined a room.
pub fn room_useroncejoined(&self, room_id: &RoomId) -> impl Iterator<Item = Result<UserId>> {
self.roomuseroncejoinedids
.scan_prefix(room_id.to_string())
.keys()
.map(|key| {
Ok(UserId::try_from(
utils::string_from_bytes(
&key?
.rsplit(|&b| b == 0xff)
.next()
.expect("rsplit always returns an element"),
)
.map_err(|_| {
Error::bad_database("User ID in room_useroncejoined is invalid unicode.")
})?,
)
.map_err(|_| Error::bad_database("User ID in room_useroncejoined is invalid."))?)
})
}
/// Returns an iterator over all invited members of a room.
pub fn room_members_invited(&self, room_id: &RoomId) -> impl Iterator<Item = Result<UserId>> {
self.roomuserid_invited
@ -1302,6 +1396,14 @@ impl Rooms {
})
}
pub fn once_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let mut userroom_id = user_id.to_string().as_bytes().to_vec();
userroom_id.push(0xff);
userroom_id.extend_from_slice(room_id.to_string().as_bytes());
Ok(self.roomuseroncejoinedids.get(userroom_id)?.is_some())
}
pub fn is_joined(&self, user_id: &UserId, room_id: &RoomId) -> Result<bool> {
let mut userroom_id = user_id.as_bytes().to_vec();
userroom_id.push(0xff);

154
src/database/rooms/edus.rs
View file

@ -14,17 +14,18 @@ use std::{
};
pub struct RoomEdus {
pub(in super::super) roomuserid_lastread: sled::Tree, // RoomUserId = Room + User
pub(in super::super) roomlatestid_roomlatest: sled::Tree, // Read Receipts, RoomLatestId = RoomId + Count + UserId
pub(in super::super) roomactiveid_userid: sled::Tree, // Typing, RoomActiveId = RoomId + TimeoutTime + Count
pub(in super::super) roomid_lastroomactiveupdate: sled::Tree, // LastRoomActiveUpdate = Count
pub(in super::super) readreceiptid_readreceipt: sled::Tree, // ReadReceiptId = RoomId + Count + UserId
pub(in super::super) roomuserid_privateread: sled::Tree, // RoomUserId = Room + User, PrivateRead = Count
pub(in super::super) roomuserid_lastprivatereadupdate: sled::Tree, // LastPrivateReadUpdate = Count
pub(in super::super) typingid_userid: sled::Tree, // TypingId = RoomId + TimeoutTime + Count
pub(in super::super) roomid_lasttypingupdate: sled::Tree, // LastRoomTypingUpdate = Count
pub(in super::super) presenceid_presence: sled::Tree, // PresenceId = RoomId + Count + UserId
pub(in super::super) userid_lastpresenceupdate: sled::Tree, // LastPresenceUpdate = Count
}
impl RoomEdus {
/// Adds an event which will be saved until a new event replaces it (e.g. read receipt).
pub fn roomlatest_update(
pub fn readreceipt_update(
&self,
user_id: &UserId,
room_id: &RoomId,
@ -36,7 +37,7 @@ impl RoomEdus {
// Remove old entry
if let Some(old) = self
.roomlatestid_roomlatest
.readreceiptid_readreceipt
.scan_prefix(&prefix)
.keys()
.rev()
@ -50,7 +51,7 @@ impl RoomEdus {
})
{
// This is the old room_latest
self.roomlatestid_roomlatest.remove(old)?;
self.readreceiptid_readreceipt.remove(old)?;
}
let mut room_latest_id = prefix;
@ -58,7 +59,7 @@ impl RoomEdus {
room_latest_id.push(0xff);
room_latest_id.extend_from_slice(&user_id.to_string().as_bytes());
self.roomlatestid_roomlatest.insert(
self.readreceiptid_readreceipt.insert(
room_latest_id,
&*serde_json::to_string(&event).expect("EduEvent::to_string always works"),
)?;
@ -67,7 +68,7 @@ impl RoomEdus {
}
/// Returns an iterator over the most recent read_receipts in a room that happened after the event with id `since`.
pub fn roomlatests_since(
pub fn readreceipts_since(
&self,
room_id: &RoomId,
since: u64,
@ -79,7 +80,7 @@ impl RoomEdus {
first_possible_edu.extend_from_slice(&(since + 1).to_be_bytes()); // +1 so we don't send the event at since
Ok(self
.roomlatestid_roomlatest
.readreceiptid_readreceipt
.range(&*first_possible_edu..)
.filter_map(|r| r.ok())
.take_while(move |(k, _)| k.starts_with(&prefix))
@ -90,9 +91,60 @@ impl RoomEdus {
}))
}
/// Sets a user as typing until the timeout timestamp is reached or roomactive_remove is
/// Sets a private read marker at `count`.
pub fn private_read_set(
&self,
room_id: &RoomId,
user_id: &UserId,
count: u64,
globals: &super::super::globals::Globals,
) -> Result<()> {
let mut key = room_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&user_id.to_string().as_bytes());
self.roomuserid_privateread
.insert(&key, &count.to_be_bytes())?;
self.roomuserid_lastprivatereadupdate
.insert(&key, &globals.next_count()?.to_be_bytes())?;
Ok(())
}
/// Returns the private read marker.
pub fn private_read_get(&self, room_id: &RoomId, user_id: &UserId) -> Result<Option<u64>> {
let mut key = room_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&user_id.to_string().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")
})?))
})
}
/// 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.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&user_id.to_string().as_bytes());
Ok(self
.roomuserid_lastprivatereadupdate
.get(&key)?
.map_or(Ok::<_, Error>(None), |bytes| {
Ok(Some(utils::u64_from_bytes(&bytes).map_err(|_| {
Error::bad_database("Count in roomuserid_lastprivatereadupdate is invalid.")
})?))
})?
.unwrap_or(0))
}
/// Sets a user as typing until the timeout timestamp is reached or roomtyping_remove is
/// called.
pub fn roomactive_add(
pub fn typing_add(
&self,
user_id: &UserId,
room_id: &RoomId,
@ -104,22 +156,22 @@ impl RoomEdus {
let count = globals.next_count()?.to_be_bytes();
let mut room_active_id = prefix;
room_active_id.extend_from_slice(&timeout.to_be_bytes());
room_active_id.push(0xff);
room_active_id.extend_from_slice(&count);
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.roomactiveid_userid
.insert(&room_active_id, &*user_id.to_string().as_bytes())?;
self.typingid_userid
.insert(&room_typing_id, &*user_id.to_string().as_bytes())?;
self.roomid_lastroomactiveupdate
self.roomid_lasttypingupdate
.insert(&room_id.to_string().as_bytes(), &count)?;
Ok(())
}
/// Removes a user from typing before the timeout is reached.
pub fn roomactive_remove(
pub fn typing_remove(
&self,
user_id: &UserId,
room_id: &RoomId,
@ -132,19 +184,19 @@ impl RoomEdus {
let mut found_outdated = false;
// Maybe there are multiple ones from calling roomactive_add multiple times
// Maybe there are multiple ones from calling roomtyping_add multiple times
for outdated_edu in self
.roomactiveid_userid
.typingid_userid
.scan_prefix(&prefix)
.filter_map(|r| r.ok())
.filter(|(_, v)| v == user_id.as_bytes())
{
self.roomactiveid_userid.remove(outdated_edu.0)?;
self.typingid_userid.remove(outdated_edu.0)?;
found_outdated = true;
}
if found_outdated {
self.roomid_lastroomactiveupdate.insert(
self.roomid_lasttypingupdate.insert(
&room_id.to_string().as_bytes(),
&globals.next_count()?.to_be_bytes(),
)?;
@ -154,7 +206,7 @@ impl RoomEdus {
}
/// Makes sure that typing events with old timestamps get removed.
fn roomactives_maintain(
fn typings_maintain(
&self,
room_id: &RoomId,
globals: &super::super::globals::Globals,
@ -168,7 +220,7 @@ impl RoomEdus {
// Find all outdated edus before inserting a new one
for outdated_edu in self
.roomactiveid_userid
.typingid_userid
.scan_prefix(&prefix)
.keys()
.map(|key| {
@ -176,21 +228,21 @@ impl RoomEdus {
Ok::<_, Error>((
key.clone(),
utils::u64_from_bytes(key.split(|&b| b == 0xff).nth(1).ok_or_else(|| {
Error::bad_database("RoomActive has invalid timestamp or delimiters.")
Error::bad_database("RoomTyping has invalid timestamp or delimiters.")
})?)
.map_err(|_| Error::bad_database("RoomActive has invalid timestamp bytes."))?,
.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.roomactiveid_userid.remove(outdated_edu.0)?;
self.typingid_userid.remove(outdated_edu.0)?;
found_outdated = true;
}
if found_outdated {
self.roomid_lastroomactiveupdate.insert(
self.roomid_lasttypingupdate.insert(
&room_id.to_string().as_bytes(),
&globals.next_count()?.to_be_bytes(),
)?;
@ -199,16 +251,16 @@ impl RoomEdus {
Ok(())
}
/// Returns an iterator over all active events (e.g. typing notifications).
pub fn last_roomactive_update(
/// Returns the count of the last typing update in this room.
pub fn last_typing_update(
&self,
room_id: &RoomId,
globals: &super::super::globals::Globals,
) -> Result<u64> {
self.roomactives_maintain(room_id, globals)?;
self.typings_maintain(room_id, globals)?;
Ok(self
.roomid_lastroomactiveupdate
.roomid_lasttypingupdate
.get(&room_id.to_string().as_bytes())?
.map_or(Ok::<_, Error>(None), |bytes| {
Ok(Some(utils::u64_from_bytes(&bytes).map_err(|_| {
@ -218,7 +270,7 @@ impl RoomEdus {
.unwrap_or(0))
}
pub fn roomactives_all(
pub fn typings_all(
&self,
room_id: &RoomId,
) -> Result<SyncEphemeralRoomEvent<ruma::events::typing::TypingEventContent>> {
@ -228,17 +280,15 @@ impl RoomEdus {
let mut user_ids = Vec::new();
for user_id in self
.roomactiveid_userid
.typingid_userid
.scan_prefix(prefix)
.values()
.map(|user_id| {
Ok::<_, Error>(
UserId::try_from(utils::string_from_bytes(&user_id?).map_err(|_| {
Error::bad_database("User ID in roomactiveid_userid is invalid unicode.")
Error::bad_database("User ID in typingid_userid is invalid unicode.")
})?)
.map_err(|_| {
Error::bad_database("User ID in roomactiveid_userid is invalid.")
})?,
.map_err(|_| Error::bad_database("User ID in typingid_userid is invalid."))?,
)
})
{
@ -250,30 +300,6 @@ impl RoomEdus {
})
}
/// Sets a private read marker at `count`.
pub fn room_read_set(&self, room_id: &RoomId, user_id: &UserId, count: u64) -> Result<()> {
let mut key = room_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&user_id.to_string().as_bytes());
self.roomuserid_lastread.insert(key, &count.to_be_bytes())?;
Ok(())
}
/// Returns the private read marker.
pub fn room_read_get(&self, room_id: &RoomId, user_id: &UserId) -> Result<Option<u64>> {
let mut key = room_id.to_string().as_bytes().to_vec();
key.push(0xff);
key.extend_from_slice(&user_id.to_string().as_bytes());
self.roomuserid_lastread.get(key)?.map_or(Ok(None), |v| {
Ok(Some(utils::u64_from_bytes(&v).map_err(|_| {
Error::bad_database("Invalid private read marker bytes")
})?))
})
}
/// 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

43
src/database/transaction_ids.rs Normal file
View file

@ -0,0 +1,43 @@
use crate::Result;
use ruma::{DeviceId, UserId};
use sled::IVec;
pub struct TransactionIds {
pub(super) userdevicetxnid_response: sled::Tree, // Response can be empty (/sendToDevice) or the event id (/send)
}
impl TransactionIds {
pub fn add_txnid(
&self,
user_id: &UserId,
device_id: &DeviceId,
txn_id: &str,
data: &[u8],
) -> Result<()> {
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(txn_id.as_bytes());
self.userdevicetxnid_response.insert(key, data)?;
Ok(())
}
pub fn existing_txnid(
&self,
user_id: &UserId,
device_id: &DeviceId,
txn_id: &str,
) -> Result<Option<IVec>> {
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(txn_id.as_bytes());
// If there's no entry, this is a new transaction
Ok(self.userdevicetxnid_response.get(key)?)
}
}